¿Qué es MCP?
MCP (Model Context Protocol) es un estándar open source desarrollado por Anthropic para permitir a los modelos de IA interactuar con herramientas externas mediante un estándar
Hasta el desarrollo del protocolo MCP, cuando queríamos que un LLM interactuara con herramientas, teníamos que crear código para poder interactuar con la herramienta, y mediante function calling
enviarle la información al LLM.

Así que mediante MCP, un LLM puede interactuar con herramientas gracias a un estándar. De esta manera si una persona crea un servidor MCP, dicho servidor puede ser reutilizado por otros con un único cliente. Si en tu aplicación desarrollas un cliente, puedes descargarte un servidor MCP desarrollado por otro, y usarlo sin problema.
Comúnmente MCP se asemeja al estándar USB. Antes del USB, cada periférico tenía un tipo de conexión diferente, unos tenían puertos serie, otros paralelo. Diferentes formatos de conectores, etc.

Con la llegada del USB, todos los periféricos se adaptaron a este estándar, por lo que con un solo conector USB en tu ordenador, puedes conectar casi cualquier periférico.
MCP tiene 7 componentes principales:
- Host: Aplicación LLM que tiene acceso a herramientas MCP.
- Servidor MCP: Servidor que realiza la comunicación con la API o herramienta a la que queremos exponer al LLM
- Cliente MCP: Cliente que se conecta al servidor MCP y realiza las peticiones
- Tool: Función que se ejecuta en el servidor MCP y que puede ser invocada por el LLM
- Resource: Recurso que se puede usar en el servidor MCP. Suelen dar al LLM acceso a recursos estáticos como archivos, bases de datos, etc.
- Resource template: Template para crear recursos dinámicos. Mediante estas plantillas, el LLM puede crear dinámicamente el recurso al que quiere acceder
- Prompt: Prompt que se usa para generar un prompt que será usado por el LLM para interactuar con el servidor MCP.
Un único host (aplicación) puede tener varios clientes. Cada cliente se conectará a un servidor MCP

FastMCP
Aunque en la documentación de MCP recomiendan instalar mcp["cli"]
, hay una librería creada por encima llamada fastmcp
, que ayuda mucho a la hora de crear servidores MCP, así que vamos a usarla
Crear entorno virtual
Para crear un servidor y un cliente MCP, vamos a crear entornos virtuales con uv
con las dependencias que vamos a necesitar
Servidor MCP
Primero creamos una carpeta para el servidor de MCP
!mkdir gitHub_MCP_server
Iniciamos el entorno uv
!cd gitHub_MCP_server && uv init .
Initialized project `github-mcp-server` at `/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server`
Lo activamos
!cd gitHub_MCP_server && uv venv
Using CPython 3.11.11Creating virtual environment at: .venvActivate with: source .venv/bin/activate
E instalamos las librerías necesarias
!cd gitHub_MCP_server && uv add anthropic fastmcp python-dotenv requests
Resolved 42 packages in 34msInstalled 40 packages in 71ms+ annotated-types==0.7.0+ anyio==4.9.0+ authlib==1.6.0+ certifi==2025.6.15+ cffi==1.17.1+ charset-normalizer==3.4.2+ click==8.2.1+ cryptography==45.0.4+ distro==1.9.0+ exceptiongroup==1.3.0+ fastmcp==2.9.0+ h11==0.16.0+ httpcore==1.0.9+ httpx==0.28.1+ httpx-sse==0.4.0+ idna==3.10+ jiter==0.10.0+ markdown-it-py==3.0.0+ mcp==1.9.4+ mdurl==0.1.2+ openapi-pydantic==0.5.1+ pycparser==2.22+ pydantic==2.11.7+ pydantic-core==2.33.2+ pydantic-settings==2.10.0+ pygments==2.19.2+ python-dotenv==1.1.1+ python-multipart==0.0.20+ requests==2.32.4+ rich==14.0.0+ shellingham==1.5.4+ sniffio==1.3.1+ sse-starlette==2.3.6+ starlette==0.47.1+ typer==0.16.0+ typing-extensions==4.14.0+ typing-inspection==0.4.1+ typing-inspection==0.4.1
Cliente MCP
Ahora creamos una carpeta donde programaremos el cliente MCP
!mkdir client_MCP
Iniciamos el entorno uv
!cd client_MCP && uv init .
Initialized project `client-mcp` at `/Users/macm1/Documents/web/portafolio/posts/client_MCP`
Lo activamos
!cd client_MCP && uv venv
Using CPython 3.11.11Creating virtual environment at: .venvActivate with: source .venv/bin/activate
Y por último, instalamos las librerías necesarias para el cliente.
!cd client_MCP && uv add anthropic fastmcp python-dotenv requests
Resolved 42 packages in 307msPrepared 5 packages in 115msInstalled 40 packages in 117ms+ annotated-types==0.7.0+ anthropic==0.55.0+ anyio==4.9.0+ authlib==1.6.0+ certifi==2025.6.15+ cffi==1.17.1+ charset-normalizer==3.4.2+ click==8.2.1+ cryptography==45.0.4+ distro==1.9.0+ exceptiongroup==1.3.0+ fastmcp==2.9.0+ h11==0.16.0+ httpcore==1.0.9+ httpx==0.28.1+ httpx-sse==0.4.0+ idna==3.10+ jiter==0.10.0+ markdown-it-py==3.0.0+ mcp==1.9.4+ mdurl==0.1.2+ openapi-pydantic==0.5.1+ pycparser==2.22+ pydantic==2.11.7+ pydantic-core==2.33.2+ pydantic-settings==2.10.0+ pygments==2.19.2+ python-dotenv==1.1.1+ python-multipart==0.0.20+ requests==2.32.4+ rich==14.0.0+ shellingham==1.5.4+ sniffio==1.3.1+ sse-starlette==2.3.6+ starlette==0.47.1+ typer==0.16.0+ typing-extensions==4.14.0+ typing-inspection==0.4.1+ typing-inspection==0.4.1
Vamos a usar Sonnet 3.5 como modelo LLM, así que creamos un archivo .env
en la carpeta del cliente con la API KEY de Claude que se puede obtener en la página keys de la API de Claude
%%writefile client_MCP/.envANTHROPIC_API_KEY="ANTHROPIC_API_KEY"
Writing client_MCP/.env
MCP básico
Escribimos el mínimo código que necesitamos para tener un servidor MCP
%%writefile gitHub_MCP_server/github_server.pyfrom mcp.server.fastmcp import FastMCP# Create an MCP servermcp = FastMCP("GitHubMCP")if __name__ == "__main__":# Initialize and run the servermcp.run(transport='stdio')
Overwriting gitHub_MCP_server/github_server.py
Como se puede ver, tenemos que crear un objeto FastMCP
y luego ejecutar el servidor con mcp.run
.
Librería con funciones para leer de GitHub
Como vamos a crear un servidor MCP para poder usar utilidades de GitHub, vamos a crear un archivo con las funciones necesarias para construir los headers necesarios para poder usar la API de GitHub.
%%writefile gitHub_MCP_server/github.pyimport osfrom dotenv import load_dotenv# Load the GitHub token from the .env fileload_dotenv()GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")# Check if the GitHub token is configuredif not GITHUB_TOKEN:print("WARNING: The GITHUB_TOKEN environment variable is not configured.")print("Requests to the GitHub API may fail due to rate limits.")print("Create a .env file in this directory with GITHUB_TOKEN='your_token_here'")raise ValueError("GITHUB_TOKEN is not configured")# Helper function to create headers for GitHub API requestsdef create_github_headers():headers = {}if GITHUB_TOKEN:headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"# GitHub recommends including a User-Agentheaders["User-Agent"] = "MCP_GitHub_Server_Example"headers["Accept"] = "application/vnd.github.v3+json" # Good practicereturn headers
Overwriting gitHub_MCP_server/github.py
Para poder construir los headers, necesitamos un token de GitHub. Para ello, vamos a personal-access-tokens y creamos un nuevo token. Lo copiamos
Ahora, creamos un .env
, donde vamos a almacenar el token de GitHub.
%%writefile gitHub_MCP_server/.envGITHUB_TOKEN = "GITHUB_TOKEN"
Overwriting gitHub_MCP_server/.env
Crear tool
de MCP para obtener una lista de issues de un repositorio de GitHub
Servidor MCP
Añadimos una función para poder listar los issues de un repositorio de GitHub. Para convertir dicha función en una tool
de MCP, usamos el decorador @mcp.tool()
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create a FastMCP servermcp = FastMCP("GitHubMCP")@mcp.tool()async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to the first 10 issues to avoid long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'Sin título')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comentarios)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Mostrando los primeros 10 issues abiertos" if len(issues_summary) == 10 else f"Mostrando todos los {len(issues_summary)} issues abiertos","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting GitHub FastMCP server...")print(f"DEBUG: Server name: {mcp.name}")print("DEBUG: Available tools: list_repository_issues")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Cliente MCP
Ahora creamos un cliente MCP para poder usar la tool
que hemos creado
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_script_path: str):"""Connect to the specified FastMCP server.Args:server_script_path: Path to the server script (Python)"""print(f"🔗 Connecting to FastMCP server: {server_script_path}")# Determine the server type based on the extensionif not server_script_path.endswith('.py'):raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")# Create FastMCP clientself.client = Client(server_script_path)# Note: FastMCP Client automatically infers transport from .py filesprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available toolstools_list = await client.list_tools()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:# Execute tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:# Convert result to string format for Clauderesult_content = str(tool_result)messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools from the FastMCP server")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <path_to_fastmcp_server>")print("📝 Example: python client.py ../MCP_github/github_server.py")sys.exit(1)server_script_path = sys.argv[1]# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_script_path)# List available tools after connectionawait client.list_available_tools()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Explicación del cliente MCP
- En
main
se comprueba que se ha pasado un argumento con el path del servidor MCP. - Se crea un objeto de la clase
FastMCPClient
con el path del servidor MCP. Al crear el objeto se ejecuta el método__init__
que crea la conexión con el LLM de Anthropic, que va a ser el LLM que va a poner el "cerebro" - Se intenta conectar con el servidor MCP llamando al método
connect_to_server
para abrir una sesión con el servidor MCP. - Se listan las
tool
s disponibles con el métodolist_available_tools
- Si se ha podido conectar, se llama al método
chat_loop
que es un bucle infinito para chatear con el LLM que se acaba de crear en el cliente. Solo se para la ejecución cuando se introducequit
,q
,exit
osalir
en el chat. - Se procesa la entrada del usuario con el método
process_query
que obtiene la lista detool
s disponibles y hace una petición al LLM con el mensaje del usuario y la lista detool
s - Si el LLM responde con texto, se devuelve el texto, que será impreso
- Si el LLM responde con
tool_use
, se obtiene el nombre de latool
, los argumentos y se crea una ID de ejecución. Se ejecuta la tool. Con el resultado de la tool, se crea un nuevo mensaje que se le manda al LLM para que lo procese y genere una respuesta, que será devuelta e impresa. - Cuando se termine la conversación, se llamará al método
cleanup
, que cerrará lo que sea necesario cerrar.
Prueba de la tool
Nos vamos a la ruta del cliente y lo ejecutamos, dándole la ruta del servidor MCP.
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully[06/28/25 09:22:09] INFO Starting MCP server 'GitHubMCP' with transport 'stdio' server.py:1246🛠️ Available tools (1):==================================================📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP server------------------------------------------------------------👤 You: Tell me de issues of repository transformers of huggingface🤔 Claude is thinking...🔧 Claude wants to use: list_repository_issues📝 Arguments: {'owner': 'huggingface', 'repo_name': 'transformers'}✅ Tool executed successfully🤖 Claude: I'll help you list the issues from the Hugging Face transformers repository. Let me use the `list_repository_issues` function with the appropriate parameters.I'll summarize the current open issues from the Hugging Face transformers repository. Here are the 10 most recent open issues:1. [#39097] Core issue about saving models with multiple shared tensor groups when dispatched2. [#39096] Pull request to fix position index in v4.52.43. [#39095] Issue with Qwen2_5_VLVisionAttention flash attention missing 'is_causal' attribute4. [#39094] Documentation improvement for PyTorch examples5. [#39093] Style change PR for lru_cache decorator6. [#39091] Compatibility issue with sentencepiece on Windows in Python 3.137. [#39090] Pull request for fixing bugs in finetune and batch inference8. [#39089] Bug report for LlavaOnevisonConfig initialization in version 4.52.49. [#39087] Documentation PR for Gemma 3n audio encoder10. [#39084] Pull request for refactoring gemma3nNote that this is showing the 10 most recent open issues, and there might be more issues in the repository. Each issue has a link where you can find more details about the specific problem or proposed changes.Would you like more specific information about any of these issues?👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Al ejecutarlo vemos
🛠️ Available tools (1):
==================================================
📋 list_repository_issues
Description: Lists open issues for a given GitHub repository.
Args:
owner: The owner of the repository (e.g., 'modelcontextprotocol')
repo_name: The name of the repository (e.g., 'python-sdk')
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
Lo que indica que el cliente MCP puede ver la tool
que hemos creado en el servidor MCP.
Después podemos ver
👤 You: Tell me de issues of repository transformers of huggingface
🤔 Claude is thinking...
🔧 Calling tool: list_repository_issues
📝 Arguments: {'owner': 'huggingface', 'repo_name': 'transformers'}
✅ Tool executed successfully
Le pedimos los issues del repositorio transformers
de huggingface
. Tras pensar un rato nos dice que va a usar la tool
list_repository_issues
con los argumentos {'owner': 'huggingface', 'repo_name': 'transformers'}
.
Por último, nos dice que la tool
se ha ejecutado correctamente.
Por último, con el resultado de ejecutar la tool
, Claude lo procesa y nos crea una respuesta con la lista de issues.
🤖 Claude: I'll help you list the issues from the Hugging Face transformers repository. Let me use the `list_repository_issues` function with the appropriate parameters.I'll summarize the current open issues from the Hugging Face transformers repository. Here are the 10 most recent open issues:
1. [#39097] Core issue about saving models with multiple shared tensor groups when dispatched
2. [#39096] Pull request to fix position index in v4.52.4
3. [#39095] Issue with Qwen2_5_VLVisionAttention flash attention missing 'is_causal' attribute
4. [#39094] Documentation improvement for PyTorch examples
5. [#39093] Style change PR for lru_cache decorator
6. [#39091] Compatibility issue with sentencepiece on Windows in Python 3.13
7. [#39090] Pull request for fixing bugs in finetune and batch inference
8. [#39089] Bug report for LlavaOnevisonConfig initialization in version 4.52.4
9. [#39087] Documentation PR for Gemma 3n audio encoder
10. [#39084] Pull request for refactoring gemma3n
Note that this is showing the 10 most recent open issues, and there might be more issues in the repository. Each issue has a link where you can find more details about the specific problem or proposed changes.
Would you like more specific information about any of these issues?
Crear el servidor MCP con más información
Servidor MCP
Antes hemos creado el servidor con mcp = FastMCP()
, pero podemos aprovechar para darle un nombre y descripción al servidor con
mcp = FastMCP(
name="GitHubMCP",
instructions="""
This server provides tools, resources and prompts to interact with the GitHub API.
"""
)
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="""This server provides tools, resources and prompts to interact with the GitHub API.""")@mcp.tool()async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The MCP context for logging.Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Filtrar tool
s mediante tags
Servidor MCP
MCP nos da la opción de poder exponer tool
s mediante tags, lo cual puede ser útil para exponer solo tool
s para depuración, para que solo las puedan usar determinados usuarios, etc.
Para ello, cuando creamos el servidor MCP indicamos los tags que queremos incluir
mcp = FastMCP(
name="GitHubMCP",
instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
include_tags=public
)
Y luego, cuando creamos la tool
podemos indicar los tags que queremos que tenga.
@mcp.tool(tags={"public", "production"})
Vamos a ver un ejemplo
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"})@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Podemos ver que hemos creado la función list_repository_issues
, que lista solo 10 issues y que tiene los tags public
y production
. Y hemos creado la función list_more_repository_issues
, que lista 100 issues de un repositorio y que tiene los tags private
y development
.
Además hemos declarado el servidor mediante
mcp = FastMCP(
name="GitHubMCP",
instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
include_tags=public
)
Por lo que el cliente solo tendrá acceso a las tool
s que tengan el tag public
, es decir, a list_repository_issues
. Solo va a poder ver una lista de 10 issues.
Prueba de los tags
Volvemos a ejecutar el cliente MCP
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully[06/28/25 09:44:55] INFO Starting MCP server 'GitHubMCP' with ]8;id=896921;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=507812;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1246\1246]8;;transport 'stdio'🛠️ Available tools (1):==================================================📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The MCP context for logging.Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP server------------------------------------------------------------👤 You:
No hace falta hacer una petición, ya que vemos lo siguiente:
🛠️ Available tools (1):
==================================================
📋 list_repository_issues
Description: Lists open issues for a given GitHub repository.
Args:
owner: The owner of the repository (e.g., 'modelcontextprotocol')
repo_name: The name of the repository (e.g., 'python-sdk')
ctx: The MCP context for logging.
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
Es decir, el cliente solo puede ver la tool
list_repository_issues
y no la tool
list_all_repository_issues
.
Cambio a private
Cambiamos include_tags
a private
para usar la tool
list_more_repository_issues
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"private"})@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Prueba del tag private
Ejecutamos otra vez el cliente con el cambio hecho
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully[06/28/25 09:51:48] INFO Starting MCP server 'GitHubMCP' with ]8;id=921531;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=418078;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1246\1246]8;;transport 'stdio'🛠️ Available tools (1):==================================================📋 list_more_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP server------------------------------------------------------------👤 You:
Al igual que antes, no hace falta hacer una petición, ya que nos muestra las tool
s disponibles y vemos que tenemos list_more_repository_issues
.
🛠️ Available tools (1):
==================================================
📋 list_more_repository_issues
Description: Lists open issues for a given GitHub repository.
Args:
owner: The owner of the repository (e.g., 'modelcontextprotocol')
repo_name: The name of the repository (e.g., 'python-sdk')
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
Vuelta a public
Volvemos a poner include_tags
a public
para usar la tool
list_repository_issues
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"})@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Excluir tool
s por tags
Al igual que antes hemos filtrado las tool
s que se pueden usar por tags, también podemos excluir tool
s por tags, para ello, a la hora de crear el servidor hay que añadir el parámetro exclude_tags
con los tags que queremos excluir.
Servidor MCP
Creamos una nueva tool
y la excluimos mediante tags
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Hemos creado la tool
first_repository_issue
, pero no la vamos a poder usar porque tiene los tags public
y first_issue
, pero a la hora de crear el servidor hemos puesto exclude_tags={"first_issue"}
.
Prueba de exclude_tags
Ejecutamos el cliente MCP
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully[06/28/25 10:00:36] INFO Starting MCP server 'GitHubMCP' with ]8;id=28274;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=529867;file:///Users/macm1/Documents/web/portafolio/posts/client_MCP/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1246\1246]8;;transport 'stdio'🛠️ Available tools (1):==================================================📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP server------------------------------------------------------------👤 You:
Vemos que no está disponible la tool
first_repository_issue
🛠️ Available tools (1):
==================================================
📋 list_repository_issues
Description: Lists open issues for a given GitHub repository.
Args:
owner: The owner of the repository (e.g., 'modelcontextprotocol')
repo_name: The name of the repository (e.g., 'python-sdk')
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
Composición de servidores
Al igual que en programación se pueden heredar clases, o construir sobre funciones ya creadas, en MCP se pueden crear sub-servidores y crear una composición de ellos.
Servidor MCP
Vamos a crear un subservidor MCP, con su propia tool
hello_world
. Después lo montamos en el servidor principal. Haciendo esto, vamos a poder usar la tool
hello_world
en el cliente que se conecte al servidor principal.
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run()
Overwriting gitHub_MCP_server/github_server.py
Prueba de la composición de servidores MCP
Ejecutamos el cliente
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:240: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)[06/28/25 10:10:58] INFO Starting MCP server 'GitHubMCP' with transport 'stdio' server.py:1246🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP server------------------------------------------------------------👤 You: Can you greeting me?🤔 Claude is thinking...🔧 Claude wants to use: sub_mcp_hello_world📝 Arguments: {}✅ Tool executed successfully🤖 Claude: I'll help you send a greeting using the `sub_mcp_hello_world` function. This function returns a simple greeting.There's your greeting! The function returned "Hello, world!"👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Podemos ver que ha aparecido la nueva tool
sub_mcp_hello_world
🛠️ Available tools (2):
==================================================
📋 sub_mcp_hello_world
Description: Returns a simple greeting.
Parameters:
📋 list_repository_issues
Description: Lists open issues for a given GitHub repository.
Args:
owner: The owner of the repository (e.g., 'modelcontextprotocol')
repo_name: The name of the repository (e.g., 'python-sdk')
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
Y cuando le pedimos que nos salude la ejecuta
👤 You: Can you greeting me?
🤔 Claude is thinking...
🔧 Claude wants to use: sub_mcp_hello_world
📝 Arguments:
✅ Tool executed successfully
🤖 Claude: I'll help you send a greeting using the `sub_mcp_hello_world` function. This function returns a simple greeting.There's your greeting! The function returned "Hello, world!"
Capa de transporte
Si al servidor MCP no le indicamos la capa de transporte, por defecto se usa stdio
. Pero podemos indicárselo mediante el parámetro transport
cuando lo ejecutamos
mcp.run(
transport="stdio"
)
Sin embargo, si el cliente y el servidor no están en el mismo ordenador, podemos usar http
como capa de transporte
Servidor MCP
En el servidor, solo tenemos que indicar que queremos usar http
como capa de transporte, el host y el puerto.
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
)
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the server, run with uv run client.py http://localhost:8000/mcpmcp.run(transport="streamable-http",host="0.0.0.0",port=8000,)
Overwriting gitHub_MCP_server/github_server.py
Cliente MCP
En el cliente, lo que hay que cambiar es que antes realizábamos la conexión, pasando el path del servidor (async def connect_to_server(self, server_script_path: str)
), mientras que ahora lo hacemos pasándole la URL del servidor (async def connect_to_server(self, server_url: str)
).
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_url: str):"""Connect to the specified FastMCP server via HTTP.Args:server_url: URL of the HTTP server (e.g., "http://localhost:8000")"""print(f"🔗 Connecting to FastMCP HTTP server: {server_url}")# Create FastMCP client for HTTP connection using SSE transportself.client = Client(server_url)# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transportprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available toolstools_list = await client.list_tools()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:# Execute tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:# Convert result to string format for Clauderesult_content = str(tool_result)messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools from the FastMCP HTTP server")print("🌐 Connected via Server-Sent Events (SSE)")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <http_server_url>")print("📝 Example: python client.py http://localhost:8000")print("📝 Note: Now connects to HTTP server instead of executing script")sys.exit(1)server_url = sys.argv[1]# Validate URL formatif not server_url.startswith(('http://', 'https://')):print("❌ Error: Server URL must start with http:// or https://")print("📝 Example: python client.py http://localhost:8000")sys.exit(1)# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_url)# List available tools after connectionawait client.list_available_tools()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Prueba del MCP por http
Para probar, primero tenemos que ejecutar el cliente para que se levante la URL y el puerto
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:240: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)DEBUG: Starting FastMCP GitHub server...DEBUG: Server name: GitHubMCP[06/28/25 10:33:36] INFO Starting MCP server 'GitHubMCP' with ]8;id=281189;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=128713;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1297\1297]8;;transport 'streamable-http' onhttp://0.0.0.0:8000/mcp/INFO: Started server process [89401]INFO: Waiting for application startup.INFO: Application startup complete.INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Ahora ejecutamos el cliente, dándole la URL del servidor MCP.
!cd client_MCP && source .venv/bin/activate && uv run client.py http://localhost:8000/mcp
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp✅ Client created successfully🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools from the FastMCP HTTP server🌐 Connected via Server-Sent Events (SSE)------------------------------------------------------------👤 You:
Vemos que se ha establecido la conexión sin problema.
Vuelta del servidor a STDIO
Volvemos a establecer STDIO
como capa de transporte del servidor
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom typing import Optionalfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headers# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"})async def list_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Argumentos excluídos
Servidor MCP
Supongamos que queremos tener trazabilidad de la ID del usuario que ha hecho una petición, tendríamos que añadir un parámetro a la tool
que se ejecute con dicha información. Pero esa información es irrelevante para el LLM, incluso por temas de seguridad, a lo mejor no queremos que dicha ID se pueda filtrar
Por lo que para que no se le pase un parámetro al LLM, a la hora de definir una tool
podemos indicar que se excluya un parámetro mediante exclude_args
.
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCPfrom github import GITHUB_TOKEN, create_github_headersUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Como se puede ver, en la tool
list_repository_issues
hemos indicado que se excluya el parámetro user_id
.
@mcp.tool(
tags=production,
exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM
)
async def list_repository_issues(owner: str, repo_name: str, user_id: int = USER_ID) -> list[dict]:
Aunque luego devolvemos "requested_by_user_id": user_id
result = {
"total_found": len(issues_summary),
"repository": f"{owner}/{repo_name}",
"note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues",
"issues": issues_summary,
"requested_by_user_id": user_id
}
Es decir, le estamos pasando la ID al LLM en el resultado. Pero en este caso, es para que a la hora de ejecutar la tool
veamos que se ha ejecutado con dicha ID.
Contexto
Podemos pasar información de contexto del servidor al cliente y viceversa.
Servidor MCP
Vamos a añadir contexto a nuestro servidor MCP.
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Hemos sustituído todos los print
s por ctx.info
. De esta manera, todas esas líneas de información ahora se pueden imprimir en el cliente si queremos.
Más adelante lo vamos a usar
Crear un resource
Vamos a crear un resource
estático a nuestro MCP
Servidor MCP
Podemos convertir una función en nuestro servidor en un resource
mediante el decorador @mcp.resource(<ENDPOINT>)
.
Un resource
es un endpoint que nos da información. Mientras que una tool
puede realizar cambios y/o acciones, un resource
sólo nos da información.
Vamos a verlo con un ejemplo.
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info() -> str:"""Returns information about the server."""return "This is the MCP GitHub server development for MaximoFN blog post"@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Como vemos, hemos creado el resource
server_info
que nos devuelve una cadena de texto con la información del servidor
Es importante ver que hemos declarado el endpoint resource://server_info
, que es obligatorio cada vez que creamos resource
s
Además, le hemos puesto la tag public
, ya que nuestro servidor MCP solo incluye las tool
s o resource
s que tengan la tag public
.
mcp = FastMCP(
name="GitHubMCP",
instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
include_tags=public,
exclude_tags=first_issue
)
Cliente MCP
Ahora tenemos que hacer que nuestro cliente pueda ver los resource
s de nuestro servidor MCP.
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_script_path: str):"""Connect to the specified FastMCP server.Args:server_script_path: Path to the server script (Python)"""print(f"🔗 Connecting to FastMCP server: {server_script_path}")# Determine the server type based on the extensionif not server_script_path.endswith('.py'):raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")# Create FastMCP clientself.client = Client(server_script_path)# Note: FastMCP Client automatically infers transport from .py filesprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools and resourcestools_list = await client.list_tools()resources_list = await client.list_resources()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resourcesif resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]claude_tools.append({"name": "read_mcp_resource","description": "Read a resource from the MCP server. Available resources: " +", ".join(resource_uris),"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read"}},"required": ["resource_uri"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools and resources from the FastMCP server")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <path_to_fastmcp_server>")print("📝 Example: python client.py ../MCP_github/github_server.py")sys.exit(1)server_script_path = sys.argv[1]# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_script_path)# List available tools and resources after connectionawait client.list_available_tools()await client.list_available_resources()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Hemos creado los métodos list_available_resources
y read_resource
para poder leer los recursos que tenemos en el servidor MCP.
Prueba del resource
Ejecutamos el cliente para poder probar el resource
que hemos creado
!cd client_MCP && source .venv/bin/activate && python client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:255: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)[06/28/25 11:09:01] INFO Starting MCP server 'GitHubMCP' with transport 'stdio' server.py:1246🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools and resources from the FastMCP server------------------------------------------------------------👤 You: Tell me the server info🤔 Claude is thinking...🔧 Claude wants to use: read_mcp_resource📝 Arguments: {'resource_uri': 'resource://server_info'}📖 Resource read successfully: resource://server_info🤖 Claude: I'll help you read the server information using the `read_mcp_resource` function with the specific resource URI for server info.The server information indicates that this is the MCP GitHub server development environment for MaximoFN blog post.👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Vemos que nos da una lista de resource
s
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: Returns information about the server.
MIME Type: text/plain
Y que cuando le pedimos la información del servidor usa el resource
server_info
que acabamos de crear.
👤 You: Tell me de server info
🤔 Claude is thinking...
🔧 Claude wants to use: read_mcp_resource
📝 Arguments: {'resource_uri': 'resource://server_info'}
📖 Resource read successfully: resource://server_info
🤖 Claude: I'll help you read the server information using the `read_mcp_resource` function with the specific resource URI for server info.The server information indicates that this is the MCP GitHub server development environment for MaximoFN blog post.
Añadir contexto al resource
Al igual que hicimos con las tool
s, podemos añadir contexto a los resource
s.
Servidor MCP
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info(ctx: Context) -> str:"""Returns information about the server."""return {"info": "This is the MCP GitHub server development for MaximoFN blog post","requested_id": ctx.request_id}@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Hemos añadido contexto al resource
server_info
para que nos devuelva la ID de la petición.
return {
"info": "This is the MCP GitHub server development for MaximoFN blog post",
"requested_id": ctx.request_id
}
Prueba del servidor con contexto en el resource
Ejecutamos el cliente
!cd client_MCP && source .venv/bin/activate && python client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:258: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)[06/28/25 11:17:41] INFO Starting MCP server 'GitHubMCP' with transport 'stdio' server.py:1246🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools and resources from the FastMCP server------------------------------------------------------------👤 You: Tell me the server info🤔 Claude is thinking...🔧 Claude wants to use: read_mcp_resource📝 Arguments: {'resource_uri': 'resource://server_info'}📖 Resource read successfully: resource://server_info🤖 Claude: I'll help you read the server information using the `read_mcp_resource` function. The server information is available at the resource URI "resource://server_info".According to the server information:- This is the MCP GitHub server development for MaximoFN blog post- The requested ID is "7"👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Como vemos nos ha dado la información del servidor y el ID de la petición.
👤 You: Tell me the server info
🤔 Claude is thinking...
🔧 Claude wants to use: read_mcp_resource
📝 Arguments: {'resource_uri': 'resource://server_info'}
📖 Resource read successfully: resource://server_info
🤖 Claude: I'll help you read the server information using the `read_mcp_resource` function. The server information is available at the resource URI "resource://server_info".According to the server information:
- This is the MCP GitHub server development for MaximoFN blog post
- The requested ID is "7"
Crear un resource template
Antes hemos creado un resource
que es un recurso estático, pero a lo mejor queremos obtener información, pero no siempre la misma, queremos que el LLM pueda decidir qué información quiere o necesitamos.
Para ello tenemos los resource template
s, que nos dan información igual que un resource
, pero de manera dinámica. En el momento de la petición se crea el resource
y se devuelve.
Servidor MCP
Crear un resource template
se hace de la misma manera que crear un resource
, es decir mediante @mcp.resource(<ENDPOINT)
, solo que ahora el endpoint es una plantilla que se rellena en el momento de la petición.
Vamos a verlo
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersimport datetimeUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info(ctx: Context) -> str:"""Returns information about the server."""return {"info": "This is the MCP GitHub server development for MaximoFN blog post","requested_id": ctx.request_id}@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:"""Returns detailed information about a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestReturns:dict: Repository information including name, description, stats, etc."""api_url = f"https://api.github.com/repos/{owner}/{repo_name}"ctx.info(f"Fetching repository information from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()repo_data = response.json()# Extract relevant repository informationrepo_info = {"name": repo_data.get("name"),"full_name": repo_data.get("full_name"),"description": repo_data.get("description"),"owner": {"login": repo_data.get("owner", {}).get("login"),"type": repo_data.get("owner", {}).get("type")},"html_url": repo_data.get("html_url"),"clone_url": repo_data.get("clone_url"),"ssh_url": repo_data.get("ssh_url"),"language": repo_data.get("language"),"size": repo_data.get("size"), # Size in KB"stargazers_count": repo_data.get("stargazers_count"),"watchers_count": repo_data.get("watchers_count"),"forks_count": repo_data.get("forks_count"),"open_issues_count": repo_data.get("open_issues_count"),"default_branch": repo_data.get("default_branch"),"created_at": repo_data.get("created_at"),"updated_at": repo_data.get("updated_at"),"pushed_at": repo_data.get("pushed_at"),"is_private": repo_data.get("private"),"is_fork": repo_data.get("fork"),"is_archived": repo_data.get("archived"),"has_issues": repo_data.get("has_issues"),"has_projects": repo_data.get("has_projects"),"has_wiki": repo_data.get("has_wiki"),"license": repo_data.get("license", {}).get("name") if repo_data.get("license") else None,"topics": repo_data.get("topics", [])}ctx.info(f"Successfully retrieved information for repository {owner}/{repo_name}")return repo_infoexcept httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 404:error_message = f"Repository {owner}/{repo_name} not found or is private."elif e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return {"error": f"GitHub API error: {e.response.status_code}","message": error_message,"repository": f"{owner}/{repo_name}"}except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return {"error": f"An unexpected error occurred: {str(e)}","repository": f"{owner}/{repo_name}"}@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Hemos creado el resource template
repository_info
, que nos da la información de un repositorio que va a determinar el LLM. Se crea la plantilla y en tiempo de ejecución se rellena con los parámetros que se le pasan.
@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})
async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:
Tanto el repositorio, como el dueño del repositorio, tienen que ser parámetros de la función.
Client MCP
Hacemos un pequeño cambio en el cliente para que el LLM entienda que hay recursos estáticos y dinámicos
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_script_path: str):"""Connect to the specified FastMCP server.Args:server_script_path: Path to the server script (Python)"""print(f"🔗 Connecting to FastMCP server: {server_script_path}")# Determine the server type based on the extensionif not server_script_path.endswith('.py'):raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")# Create FastMCP clientself.client = Client(server_script_path)# Note: FastMCP Client automatically infers transport from .py filesprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools and resourcestools_list = await client.list_tools()resources_list = await client.list_resources()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resources (including template resources)resource_description = "Read a resource from the MCP server. "if resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]resource_description += f"Available static resources: {', '.join(resource_uris)}. "resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."claude_tools.append({"name": "read_mcp_resource","description": resource_description,"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"}},"required": ["resource_uri"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools and resources from the FastMCP server")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <path_to_fastmcp_server>")print("📝 Example: python client.py ../MCP_github/github_server.py")sys.exit(1)server_script_path = sys.argv[1]# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_script_path)# List available tools and resources after connectionawait client.list_available_tools()await client.list_available_resources()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Como vemos le decimos "description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"
claude_tools.append({
"name": "read_mcp_resource",
"description": "Read a resource from the MCP server. Available resources: " +
", ".join(resource_uris),
"input_schema": {
"type": "object",
"properties": {
"resource_uri": {
"type": "string",
"description": "URI of the resource to read"
}
},
"required": ["resource_uri"]
}
})
Prueba del resource template
Ejecutamos el cliente
!cd client_MCP && source .venv/bin/activate && python client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools and resources from the FastMCP server------------------------------------------------------------👤 You: Can you read the resource github://repo/facebook/react for detailed information about the repository?🤔 Claude is thinking...🔧 Claude wants to use: read_mcp_resource📝 Arguments: {'resource_uri': 'github://repo/facebook/react'}📖 Resource read successfully: github://repo/facebook/react🤖 Claude: I'll help you read the GitHub repository information for Facebook's React using the `read_mcp_resource` function.Based on the repository information retrieved, here are the key details about the Facebook React repository:1. Description: The library for web and native user interfaces2. Owner: Facebook (Organization)3. Language: JavaScript4. Repository Statistics:- Stars: 236,803- Forks: 48,815- Open Issues: 999- Watchers: 236,8035. Important Dates:- Created: May 24, 2013- Last Updated: June 28, 2025- Last Push: June 27, 20256. Repository Features:- Public repository (not private)- Not a fork- Not archived- Has issues enabled- Projects disabled- Wiki disabled7. License: MIT License8. Topics/Tags:- declarative- frontend- javascript- library- react- uiThe repository can be accessed via:- HTTPS: https://github.com/facebook/react- SSH: git@github.com:facebook/react.gitThis is one of the most popular repositories on GitHub, as evidenced by its high number of stars and forks, and it remains actively maintained with regular updates.👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Le pedimos la información de un repositorio, usa el resource template
repository_info
y nos da la información del repositorio.
👤 You: Can you read the resource github://repo/facebook/react for detailed information about the repository?
🤔 Claude is thinking...
🔧 Claude wants to use: read_mcp_resource
📝 Arguments: {'resource_uri': 'github://repo/facebook/react'}
📖 Resource read successfully: github://repo/facebook/react
🤖 Claude: I'll help you read the GitHub repository information for Facebook's React using the `read_mcp_resource` function.Based on the repository information retrieved, here are the key details about the Facebook React repository:
1. Description: The library for web and native user interfaces
2. Owner: Facebook (Organization)
3. Language: JavaScript
4. Repository Statistics:
- Stars: 236,803
- Forks: 48,815
- Open Issues: 999
- Watchers: 236,803
5. Important Dates:
- Created: May 24, 2013
- Last Updated: June 28, 2025
- Last Push: June 27, 2025
6. Repository Features:
- Public repository (not private)
- Not a fork
- Not archived
- Has issues enabled
- Projects disabled
- Wiki disabled
7. License: MIT License
8. Topics/Tags:
- declarative
- frontend
- javascript
- library
- react
- ui
The repository can be accessed via:
- HTTPS: https://github.com/facebook/react
- SSH: git@github.com:facebook/react.git
This is one of the most popular repositories on GitHub, as evidenced by its high number of stars and forks, and it remains actively maintained with regular updates.
Crear un prompt
Otra de las herramientas que nos ofrece MCP es pedirle al LLM que nos cree un prompt
para usarlo en una petición.
Servidor MCP
Creamos un prompt
en nuestro servidor, para ello usamos el decorador @mcp.prompt
y le pasamos el nombre del prompt, la descripción y la etiqueta public
, porque habíamos definido nuestro servidor haciendo que solo incluya las tool
s, los resource
s y los prompt
s con la etiqueta public
.
@mcp.prompt(
name="generate_issues_prompt",
description="Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.",
tags=public
)
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersimport datetimeUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info(ctx: Context) -> str:"""Returns information about the server."""return {"info": "This is the MCP GitHub server development for MaximoFN blog post","requested_id": ctx.request_id}# Use: ¿Puedes leer el resource github://repo/facebook/react para obtener información detallada del repositorio?@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:"""Returns detailed information about a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestReturns:dict: Repository information including name, description, stats, etc."""api_url = f"https://api.github.com/repos/{owner}/{repo_name}"ctx.info(f"Fetching repository information from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()repo_data = response.json()# Extract relevant repository informationrepo_info = {"name": repo_data.get("name"),"full_name": repo_data.get("full_name"),"description": repo_data.get("description"),"owner": {"login": repo_data.get("owner", {}).get("login"),"type": repo_data.get("owner", {}).get("type")},"html_url": repo_data.get("html_url"),"clone_url": repo_data.get("clone_url"),"ssh_url": repo_data.get("ssh_url"),"language": repo_data.get("language"),"size": repo_data.get("size"), # Size in KB"stargazers_count": repo_data.get("stargazers_count"),"watchers_count": repo_data.get("watchers_count"),"forks_count": repo_data.get("forks_count"),"open_issues_count": repo_data.get("open_issues_count"),"default_branch": repo_data.get("default_branch"),"created_at": repo_data.get("created_at"),"updated_at": repo_data.get("updated_at"),"pushed_at": repo_data.get("pushed_at"),"is_private": repo_data.get("private"),"is_fork": repo_data.get("fork"),"is_archived": repo_data.get("archived"),"has_issues": repo_data.get("has_issues"),"has_projects": repo_data.get("has_projects"),"has_wiki": repo_data.get("has_wiki"),"license": repo_data.get("license", {}).get("name") if repo_data.get("license") else None,"topics": repo_data.get("topics", [])}ctx.info(f"Successfully retrieved information for repository {owner}/{repo_name}")return repo_infoexcept httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 404:error_message = f"Repository {owner}/{repo_name} not found or is private."elif e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return {"error": f"GitHub API error: {e.response.status_code}","message": error_message,"repository": f"{owner}/{repo_name}"}except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return {"error": f"An unexpected error occurred: {str(e)}","repository": f"{owner}/{repo_name}"}@mcp.prompt(name="generate_issues_prompt",description="Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.",tags={"public"})def generate_issues_prompt(owner: str, repo_name: str) -> str:"""Generates a structured prompt for asking about GitHub repository issues.This prompt template helps users formulate clear questions about repository issuesand can be used as a starting point for issue analysis or research.Args:owner: Repository owner (e.g., 'huggingface', 'microsoft')repo_name: Repository name (e.g., 'transformers', 'vscode')Returns:A formatted prompt asking about repository issues"""return f"""Please provide information about the open issues in the repository {owner}/{repo_name}.I'm interested in:- Current open issues and their status- Recent issue trends and patterns- Common issue categories or topics- Any critical or high-priority issuesRepository: {owner}/{repo_name}"""@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the servermcp.run(transport="stdio")
Overwriting gitHub_MCP_server/github_server.py
Cliente MCP
Modificamos nuestro cliente para poder usar el prompt
que hemos creado en nuestro servidor.
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_script_path: str):"""Connect to the specified FastMCP server.Args:server_script_path: Path to the server script (Python)"""print(f"🔗 Connecting to FastMCP server: {server_script_path}")# Determine the server type based on the extensionif not server_script_path.endswith('.py'):raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")# Create FastMCP clientself.client = Client(server_script_path)# Note: FastMCP Client automatically infers transport from .py filesprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def list_available_prompts(self):"""List available prompts in the FastMCP server."""try:# Get list of prompts from the server using FastMCP contextasync with self.client as client:prompts = await client.list_prompts()if prompts:print(f" 💭 Available prompts ({len(prompts)}):")print("=" * 50)for prompt in prompts:print(f"🎯 {prompt.name}")if prompt.description:print(f" Description: {prompt.description}")# Show parameters if availableif hasattr(prompt, 'arguments') and prompt.arguments:params = []for arg in prompt.arguments:param_info = f"{arg.name}: {arg.description or 'No description'}"if arg.required:param_info += " (required)"params.append(param_info)print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No prompts found in the server")except Exception as e:print(f"❌ Error listing prompts: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def get_prompt(self, prompt_name: str, prompt_args: dict = None):"""Get/call a specific prompt from the server.Args:prompt_name: Name of the prompt to callprompt_args: Arguments for the prompt (if any)Returns:str: Generated prompt content"""try:async with self.client as client:if prompt_args:result = await client.get_prompt(prompt_name, prompt_args)else:result = await client.get_prompt(prompt_name)# Extract the prompt text from the responseif hasattr(result, 'messages') and result.messages:# FastMCP returns prompts as message objectsreturn ' '.join([msg.content.text for msg in result.messages if hasattr(msg.content, 'text')])elif hasattr(result, 'content'):return str(result.content)else:return str(result)except Exception as e:print(f"❌ Error getting prompt {prompt_name}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools, resources, and promptstools_list = await client.list_tools()resources_list = await client.list_resources()prompts_list = await client.list_prompts()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resources (including template resources)resource_description = "Read a resource from the MCP server. "if resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]resource_description += f"Available static resources: {', '.join(resource_uris)}. "resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."claude_tools.append({"name": "read_mcp_resource","description": resource_description,"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"}},"required": ["resource_uri"]}})# Add a special tool for using promptsprompt_description = "Generate specialized prompts from the MCP server. Use this when users want to: "prompt_description += "- Create well-structured questions about repositories "prompt_description += "- Get help formulating prompts for specific tasks "prompt_description += "- Generate template questions for analysis "if prompts_list:prompt_names = [p.name for p in prompts_list]prompt_description += f" Available prompts: {', '.join(prompt_names)} "prompt_description += "- generate_issues_prompt: Creates structured questions about GitHub repository issues"prompt_description += " IMPORTANT: Use prompts when users explicitly ask for help creating questions or prompts, or when they want to formulate better questions about repositories."claude_tools.append({"name": "use_mcp_prompt","description": prompt_description,"input_schema": {"type": "object","properties": {"prompt_name": {"type": "string","description": "Name of the prompt to use. Available: 'generate_issues_prompt'"},"prompt_args": {"type": "object","description": "Arguments for the prompt. For generate_issues_prompt: {opening_brace}'owner': 'repo-owner', 'repo_name': 'repo-name'{closing_brace}","properties": {"owner": {"type": "string","description": "Repository owner (e.g., 'huggingface', 'microsoft')"},"repo_name": {"type": "string","description": "Repository name (e.g., 'transformers', 'vscode')"}}}},"required": ["prompt_name"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelif tool_name == "use_mcp_prompt":# Handle prompt usageprompt_name = tool_args.get("prompt_name")prompt_args = tool_args.get("prompt_args", {})if prompt_name:tool_result = await self.get_prompt(prompt_name, prompt_args)print(f"💭 Prompt '{prompt_name}' generated successfully")result_content = str(tool_result) if tool_result else "Error generating prompt"else:tool_result = "Error: No prompt name provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools, resources, and prompts from the FastMCP server")print()print("💭 PROMPT Examples:")print(" • 'Generate a prompt for asking about issues in facebook/react'")print(" • 'Help me create a good question about microsoft/vscode issues'")print(" • 'I need a structured prompt for analyzing tensorflow/tensorflow'")print()print("🔧 DIRECT Examples:")print(" • 'Show me the issues in huggingface/transformers'")print(" • 'Get repository info for github://repo/google/chrome'")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <path_to_fastmcp_server>")print("📝 Example: python client.py ../MCP_github/github_server.py")sys.exit(1)server_script_path = sys.argv[1]# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_script_path)# List available tools, resources, and prompts after connectionawait client.list_available_tools()await client.list_available_resources()await client.list_available_prompts()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Hemos creado las funciones list_available_prompts
y get_prompt
para listar los prompt
s disponibles y obtener un prompt
específico.
Prueba del prompt
Ejecutamos el cliente
!cd client_MCP && source .venv/bin/activate && python client.py ../gitHub_MCP_server/github_server.py
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py✅ Client created successfully🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain💭 Available prompts (1):==================================================🎯 generate_issues_promptDescription: Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.Parameters: owner: No description (required), repo_name: No description (required)🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools, resources, and prompts from the FastMCP server💭 PROMPT Examples:• 'Generate a prompt for asking about issues in facebook/react'• 'Help me create a good question about microsoft/vscode issues'• 'I need a structured prompt for analyzing tensorflow/tensorflow'🔧 DIRECT Examples:• 'Show me the issues in huggingface/transformers'• 'Get repository info for github://repo/google/chrome'------------------------------------------------------------👤 You: Can you create a prompt to view issues from the huggingface transformers repository?🤔 Claude is thinking...🔧 Claude wants to use: use_mcp_prompt📝 Arguments: {'prompt_name': 'generate_issues_prompt', 'prompt_args': {'owner': 'huggingface', 'repo_name': 'transformers'}}💭 Prompt 'generate_issues_prompt' generated successfully🤖 Claude: I'll help you generate a structured prompt for viewing issues from the Hugging Face Transformers repository using the `use_mcp_prompt` function with the `generate_issues_prompt` prompt type. I have all the required information from your request:- owner: `huggingface'- repo_name: 'transformers'I've generated a structured prompt that you can use to analyze issues in the Hugging Face Transformers repository. This prompt is designed to help you get comprehensive information about the repository's issues, including their current status, trends, categories, and priorities.Would you like me to actually fetch the current issues from the repository using this prompt? If so, I can use the `list_repository_issues` function to get that information for you.👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Vemos que nos da una lista de los prompt
s disponibles.
💭 Available prompts (1):
==================================================
🎯 generate_issues_prompt
Description: Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.
Parameters: owner: No description (required), repo_name: No description (required)
🤖 FastMCP client started. Write 'quit', 'q', 'exit', 'salir' to exit.
💬 You can ask questions about GitHub repositories!
📚 The client can use tools, resources, and prompts from the FastMCP server
💭 PROMPT Examples:
• 'Generate a prompt for asking about issues in facebook/react'
• 'Help me create a good question about microsoft/vscode issues'
• 'I need a structured prompt for analyzing tensorflow/tensorflow'
🔧 DIRECT Examples:
• 'Show me the issues in huggingface/transformers'
• 'Get repository info for github://repo/google/chrome'
Y que cuando le pedimos un prompt
nos da el prompt
generado.
👤 You: Can you create a prompt to view issues from the huggingface transformers repository?
🤔 Claude is thinking...
🔧 Claude wants to use: use_mcp_prompt
📝 Arguments: {'prompt_name': 'generate_issues_prompt', 'prompt_args': {'owner': 'huggingface', 'repo_name': 'transformers'}}
💭 Prompt 'generate_issues_prompt' generated successfully
🤖 Claude: I'll help you generate a structured prompt for viewing issues from the Hugging Face Transformers repository using the `use_mcp_prompt` function with the `generate_issues_prompt` prompt type. I have all the required information from your request:
- owner: "huggingface"
- repo_name: "transformers"I've generated a structured prompt that you can use to analyze issues in the Hugging Face Transformers repository. This prompt is designed to help you get comprehensive information about the repository's issues, including their current status, trends, categories, and priorities.
Would you like me to actually fetch the current issues from the repository using this prompt? If so, I can use the `list_repository_issues` function to get that information for you.
Vuelta a HTTP
Volvemos a configurar http
como capa de transporte para las dos últimas cosas que vamos a ver
Servidor MCP
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom github import GITHUB_TOKEN, create_github_headersimport datetimeUSER_ID = 1234567890# Create FastMCP servermcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"})sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info(ctx: Context) -> str:"""Returns information about the server."""return {"info": "This is the MCP GitHub server development for MaximoFN blog post","requested_id": ctx.request_id}# Use: ¿Puedes leer el resource github://repo/facebook/react para obtener información detallada del repositorio?@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:"""Returns detailed information about a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestReturns:dict: Repository information including name, description, stats, etc."""api_url = f"https://api.github.com/repos/{owner}/{repo_name}"ctx.info(f"Fetching repository information from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()repo_data = response.json()# Extract relevant repository informationrepo_info = {"name": repo_data.get("name"),"full_name": repo_data.get("full_name"),"description": repo_data.get("description"),"owner": {"login": repo_data.get("owner", {}).get("login"),"type": repo_data.get("owner", {}).get("type")},"html_url": repo_data.get("html_url"),"clone_url": repo_data.get("clone_url"),"ssh_url": repo_data.get("ssh_url"),"language": repo_data.get("language"),"size": repo_data.get("size"), # Size in KB"stargazers_count": repo_data.get("stargazers_count"),"watchers_count": repo_data.get("watchers_count"),"forks_count": repo_data.get("forks_count"),"open_issues_count": repo_data.get("open_issues_count"),"default_branch": repo_data.get("default_branch"),"created_at": repo_data.get("created_at"),"updated_at": repo_data.get("updated_at"),"pushed_at": repo_data.get("pushed_at"),"is_private": repo_data.get("private"),"is_fork": repo_data.get("fork"),"is_archived": repo_data.get("archived"),"has_issues": repo_data.get("has_issues"),"has_projects": repo_data.get("has_projects"),"has_wiki": repo_data.get("has_wiki"),"license": repo_data.get("license", {}).get("name") if repo_data.get("license") else None,"topics": repo_data.get("topics", [])}ctx.info(f"Successfully retrieved information for repository {owner}/{repo_name}")return repo_infoexcept httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 404:error_message = f"Repository {owner}/{repo_name} not found or is private."elif e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return {"error": f"GitHub API error: {e.response.status_code}","message": error_message,"repository": f"{owner}/{repo_name}"}except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return {"error": f"An unexpected error occurred: {str(e)}","repository": f"{owner}/{repo_name}"}@mcp.prompt(name="generate_issues_prompt",description="Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.",tags={"public"})def generate_issues_prompt(owner: str, repo_name: str) -> str:"""Generates a structured prompt for asking about GitHub repository issues.This prompt template helps users formulate clear questions about repository issuesand can be used as a starting point for issue analysis or research.Args:owner: Repository owner (e.g., 'huggingface', 'microsoft')repo_name: Repository name (e.g., 'transformers', 'vscode')Returns:A formatted prompt asking about repository issues"""return f"""Please provide information about the open issues in the repository {owner}/{repo_name}.I'm interested in:- Current open issues and their status- Recent issue trends and patterns- Common issue categories or topics- Any critical or high-priority issuesRepository: {owner}/{repo_name}"""@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the server, run with uv run client.py http://localhost:8000/mcpmcp.run(transport="streamable-http",host="0.0.0.0",port=8000,)
Overwriting gitHub_MCP_server/github_server.py
Cliente MCP
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Client# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_url: str):"""Connect to the specified FastMCP server via HTTP.Args:server_url: URL of the HTTP server (e.g., "http://localhost:8000/mcp")"""print(f"🔗 Connecting to FastMCP HTTP server: {server_url}")# Create FastMCP client for HTTP connection using SSE transportself.client = Client(server_url)# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transportprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools and resourcestools_list = await client.list_tools()resources_list = await client.list_resources()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resources (including template resources)resource_description = "Read a resource from the MCP server. "if resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]resource_description += f"Available static resources: {', '.join(resource_uris)}. "resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."claude_tools.append({"name": "read_mcp_resource","description": resource_description,"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"}},"required": ["resource_uri"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools and resources from the FastMCP server")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) != 2:print("❌ Usage: python client.py <http_server_url>")print("📝 Example: python client.py http://localhost:8000/mcp")print("📝 Note: Now connects to HTTP server instead of executing script")sys.exit(1)server_url = sys.argv[1]# Validate URL formatif not server_url.startswith(('http://', 'https://')):print("❌ Error: Server URL must start with http:// or https://")print("📝 Example: python client.py http://localhost:8000")sys.exit(1)# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_url)# List available tools and resources after connectionawait client.list_available_tools()await client.list_available_resources()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Autenticación
Si queremos crear un servidor MCP al que solo se pueda conectar determinados clientes, podemos añadir autenticación
Servidor MCP
Creamos el servidor con autenticación
%%writefile gitHub_MCP_server/github_server.pyimport httpxfrom fastmcp import FastMCP, Contextfrom fastmcp.server.auth import BearerAuthProviderfrom fastmcp.server.auth.providers.bearer import RSAKeyPairfrom fastmcp.server.dependencies import get_access_token, AccessTokenfrom github import GITHUB_TOKEN, create_github_headersimport datetimeUSER_ID = 1234567890# Generate RSA key pair for development and testingprint("🔐 Generating RSA key pair for authentication...")key_pair = RSAKeyPair.generate()# Configure Bearer authentication providerauth_provider = BearerAuthProvider(public_key=key_pair.public_key,issuer="https://github-mcp.maxfn.dev",audience="github-mcp-server",required_scopes=["github:read"] # Global scope required for all requests)# Generate a test token for developmentdevelopment_token = key_pair.create_token(subject="dev-user-maxfn",issuer="https://github-mcp.maxfn.dev",audience="github-mcp-server",scopes=["github:read", "github:write"],expires_in_seconds=3600 * 24 # Token is valid for 24 hours)print(f"🎫 Development token generated:")print(f" {development_token}")print("💡 Use this token in the client to authenticate")print("-" * 60)# Create FastMCP server with authenticationmcp = FastMCP(name="GitHubMCP",instructions="This server provides tools, resources and prompts to interact with the GitHub API.",include_tags={"public"},exclude_tags={"first_issue"},auth=auth_provider # Add authentication to the server)sub_mcp = FastMCP(name="SubMCP",)@mcp.tool(tags={"public", "production"},exclude_args=["user_id"], # user_id has to be injected by server, not provided by LLM)async def list_repository_issues(owner: str, repo_name: str, ctx: Context, user_id: int = USER_ID) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 10 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=10"ctx.info(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:ctx.info("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})ctx.info(f"Found {len(issues_summary)} open issues.")# Get authenticated access token informationtry:access_token: AccessToken = get_access_token()authenticated_user = access_token.client_iduser_scopes = access_token.scopesctx.info(f"Request authenticated for user: {authenticated_user} with scopes: {user_scopes}")except Exception as e:authenticated_user = "unknown"user_scopes = []ctx.warning(f"Could not get access token info: {e}")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary,"requested_by_user_id": user_id,"authenticated_user": authenticated_user,"user_scopes": user_scopes}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"private", "development"})async def list_more_repository_issues(owner: str, repo_name: str) -> list[dict]:"""Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list of dictionaries, each containing information about an issue"""# Limit to first 100 issues to avoid very long responsesapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&per_page=100"print(f"Fetching issues from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No open issues found for this repository.")return [{"message": "No open issues found for this repository."}]issues_summary = []for issue in issues_data:# Create a more concise summarysummary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"if issue.get('comments', 0) > 0:summary += f" ({issue.get('comments')} comments)"issues_summary.append({"number": issue.get("number"),"title": issue.get("title"),"user": issue.get("user", {}).get("login"),"url": issue.get("html_url"),"comments": issue.get("comments"),"summary": summary})print(f"Found {len(issues_summary)} open issues.")# Add context informationresult = {"total_found": len(issues_summary),"repository": f"{owner}/{repo_name}","note": "Showing first 10 open issues" if len(issues_summary) == 10 else f"Showing all {len(issues_summary)} open issues","issues": issues_summary}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.tool(tags={"public", "first_issue"})async def first_repository_issue(owner: str, repo_name: str) -> list[dict]:"""Gets the first issue ever created in a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')Returns:list[dict]: A list containing information about the first issue created"""# Get the first issue by sorting by creation date in ascending orderapi_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&sort=created&direction=asc&per_page=1"print(f"Fetching first issue from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()issues_data = response.json()if not issues_data:print("No issues found for this repository.")return [{"message": "No issues found for this repository."}]first_issue = issues_data[0]# Create a detailed summary of the first issuesummary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"if first_issue.get('comments', 0) > 0:summary += f" ({first_issue.get('comments')} comments)"issue_info = {"number": first_issue.get("number"),"title": first_issue.get("title"),"user": first_issue.get("user", {}).get("login"),"url": first_issue.get("html_url"),"state": first_issue.get("state"),"comments": first_issue.get("comments"),"created_at": first_issue.get("created_at"),"updated_at": first_issue.get("updated_at"),"body": first_issue.get("body", "")[:500] + "..." if len(first_issue.get("body", "")) > 500 else first_issue.get("body", ""),"summary": summary}print(f"Found first issue: #{first_issue.get('number')} created on {first_issue.get('created_at')}")# Add context informationresult = {"repository": f"{owner}/{repo_name}","note": "This is the very first issue created in this repository","first_issue": issue_info}return [result]except httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"print(f"GitHub API error: {e.response.status_code}. {error_message}")return [{"error": f"GitHub API error: {e.response.status_code}","message": error_message}]except Exception as e:print(f"An unexpected error occurred: {str(e)}")return [{"error": f"An unexpected error occurred: {str(e)}"}]@mcp.resource("resource://server_info", tags={"public"})def server_info(ctx: Context) -> str:"""Returns information about the server."""return {"info": "This is the MCP GitHub server development for MaximoFN blog post","requested_id": ctx.request_id}@mcp.resource("github://repo/{owner}/{repo_name}", tags={"public"})async def repository_info(owner: str, repo_name: str, ctx: Context) -> dict:"""Returns detailed information about a GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestReturns:dict: Repository information including name, description, stats, etc."""api_url = f"https://api.github.com/repos/{owner}/{repo_name}"ctx.info(f"Fetching repository information from {api_url}...")async with httpx.AsyncClient() as client:try:response = await client.get(api_url, headers=create_github_headers())response.raise_for_status()repo_data = response.json()# Extract relevant repository informationrepo_info = {"name": repo_data.get("name"),"full_name": repo_data.get("full_name"),"description": repo_data.get("description"),"owner": {"login": repo_data.get("owner", {}).get("login"),"type": repo_data.get("owner", {}).get("type")},"html_url": repo_data.get("html_url"),"clone_url": repo_data.get("clone_url"),"ssh_url": repo_data.get("ssh_url"),"language": repo_data.get("language"),"size": repo_data.get("size"), # Size in KB"stargazers_count": repo_data.get("stargazers_count"),"watchers_count": repo_data.get("watchers_count"),"forks_count": repo_data.get("forks_count"),"open_issues_count": repo_data.get("open_issues_count"),"default_branch": repo_data.get("default_branch"),"created_at": repo_data.get("created_at"),"updated_at": repo_data.get("updated_at"),"pushed_at": repo_data.get("pushed_at"),"is_private": repo_data.get("private"),"is_fork": repo_data.get("fork"),"is_archived": repo_data.get("archived"),"has_issues": repo_data.get("has_issues"),"has_projects": repo_data.get("has_projects"),"has_wiki": repo_data.get("has_wiki"),"license": repo_data.get("license", {}).get("name") if repo_data.get("license") else None,"topics": repo_data.get("topics", [])}ctx.info(f"Successfully retrieved information for repository {owner}/{repo_name}")return repo_infoexcept httpx.HTTPStatusError as e:error_message = e.response.json().get("message", "No additional message from API.")if e.response.status_code == 404:error_message = f"Repository {owner}/{repo_name} not found or is private."elif e.response.status_code == 403 and GITHUB_TOKEN:error_message += " (Rate limit with token or token lacks permissions?)"elif e.response.status_code == 403 and not GITHUB_TOKEN:error_message += " (Rate limit without token. Consider creating a .env file with GITHUB_TOKEN.)"ctx.error(f"GitHub API error: {e.response.status_code}. {error_message}")return {"error": f"GitHub API error: {e.response.status_code}","message": error_message,"repository": f"{owner}/{repo_name}"}except Exception as e:ctx.error(f"An unexpected error occurred: {str(e)}")return {"error": f"An unexpected error occurred: {str(e)}","repository": f"{owner}/{repo_name}"}@mcp.prompt(name="generate_issues_prompt",description="Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.",tags={"public"})def generate_issues_prompt(owner: str, repo_name: str) -> str:"""Generates a structured prompt for asking about GitHub repository issues.This prompt template helps users formulate clear questions about repository issuesand can be used as a starting point for issue analysis or research.Args:owner: Repository owner (e.g., 'huggingface', 'microsoft')repo_name: Repository name (e.g., 'transformers', 'vscode')Returns:A formatted prompt asking about repository issues"""return f"""Please provide information about the open issues in the repository {owner}/{repo_name}.I'm interested in:- Current open issues and their status- Recent issue trends and patterns- Common issue categories or topics- Any critical or high-priority issuesRepository: {owner}/{repo_name}"""@sub_mcp.tool(tags={"public"})def hello_world() -> str:"""Returns a simple greeting."""return "Hello, world!"mcp.mount("sub_mcp", sub_mcp)if __name__ == "__main__":print("DEBUG: Starting FastMCP GitHub server...")print(f"DEBUG: Server name: {mcp.name}")# Initialize and run the server, run with uv run client.py http://localhost:8000/mcp# 1. Run server with uv run github_server.py. It gives you a token to use in the client.py# 2. Run client.py with the token you got from the server.py - uv run client.py http://localhost:8000/mcp <your_bearer_token>mcp.run(transport="streamable-http",host="0.0.0.0",port=8000,)
Overwriting gitHub_MCP_server/github_server.py
Creamos un proveedor de autenticación para el servidor y un token de desarrollo temporal
# Generate RSA key pair for development and testing
print("🔐 Generating RSA key pair for authentication...")
key_pair = RSAKeyPair.generate()
# Configure Bearer authentication provider
auth_provider = BearerAuthProvider(
public_key=key_pair.public_key,
issuer="https://github-mcp.maxfn.dev",
audience="github-mcp-server",
required_scopes=["github:read"] # Global scope required for all requests
)
# Generate a test token for development
development_token = key_pair.create_token(
subject="dev-user-maxfn",
issuer="https://github-mcp.maxfn.dev",
audience="github-mcp-server",
scopes=["github:read", "github:write"],
expires_in_seconds=3600 * 24 # Token is valid for 24 hours
)
print(f"🎫 Development token generated:")
print(f" {development_token}")
print("💡 Use this token in the client to authenticate")
print("-" * 60)
Lo usamos al crear el servidor MCP
mcp = FastMCP(
name="GitHubMCP",
instructions="This server provides tools, resources and prompts to interact with the GitHub API.",
include_tags=public,
exclude_tags=first_issue,
auth=auth_provider # Add authentication to the server
)
Cliente MCP
Creamos el cliente MCP con autenticación
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Clientfrom fastmcp.client.auth import BearerAuth# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_url: str, auth_token: str = None):"""Connect to the specified FastMCP server via HTTP with optional authentication.Args:server_url: URL of the HTTP server (e.g., "http://localhost:8000/mcp")auth_token: Bearer token for authentication (optional)"""print(f"🔗 Connecting to FastMCP HTTP server: {server_url}")# Create authentication if token is providedauth = Noneif auth_token:auth = BearerAuth(token=auth_token)print("🔐 Using Bearer token authentication")else:print("⚠️ No authentication token provided - connecting without auth")# Create FastMCP client for HTTP connection using SSE transportself.client = Client(server_url, auth=auth)# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transportprint("✅ Client created successfully")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def list_available_prompts(self):"""List available prompts in the FastMCP server."""try:# Get list of prompts from the server using FastMCP contextasync with self.client as client:prompts = await client.list_prompts()if prompts:print(f" 💭 Available prompts ({len(prompts)}):")print("=" * 50)for prompt in prompts:print(f"🎯 {prompt.name}")if prompt.description:print(f" Description: {prompt.description}")# Show parameters if availableif hasattr(prompt, 'arguments') and prompt.arguments:params = []for arg in prompt.arguments:param_info = f"{arg.name}: {arg.description or 'No description'}"if arg.required:param_info += " (required)"params.append(param_info)print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No prompts found in the server")except Exception as e:print(f"❌ Error listing prompts: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def get_prompt(self, prompt_name: str, prompt_args: dict = None):"""Get/call a specific prompt from the server.Args:prompt_name: Name of the prompt to callprompt_args: Arguments for the prompt (if any)Returns:str: Generated prompt content"""try:async with self.client as client:if prompt_args:result = await client.get_prompt(prompt_name, prompt_args)else:result = await client.get_prompt(prompt_name)# Extract the prompt text from the responseif hasattr(result, 'messages') and result.messages:# FastMCP returns prompts as message objectsreturn ' '.join([msg.content.text for msg in result.messages if hasattr(msg.content, 'text')])elif hasattr(result, 'content'):return str(result.content)else:return str(result)except Exception as e:print(f"❌ Error getting prompt {prompt_name}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools and resourcestools_list = await client.list_tools()resources_list = await client.list_resources()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resources (including template resources)resource_description = "Read a resource from the MCP server. "if resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]resource_description += f"Available static resources: {', '.join(resource_uris)}. "resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."claude_tools.append({"name": "read_mcp_resource","description": resource_description,"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"}},"required": ["resource_uri"]}})# Add a special tool for using promptsprompt_description = "Generate specialized prompts from the MCP server. Use this when users want to: "prompt_description += "- Create well-structured questions about repositories "prompt_description += "- Get help formulating prompts for specific tasks "prompt_description += "- Generate template questions for analysis "if prompts_list:prompt_names = [p.name for p in prompts_list]prompt_description += f" Available prompts: {', '.join(prompt_names)} "prompt_description += "- generate_issues_prompt: Creates structured questions about GitHub repository issues"prompt_description += " IMPORTANT: Use prompts when users explicitly ask for help creating questions or prompts, or when they want to formulate better questions about repositories."claude_tools.append({"name": "use_mcp_prompt","description": prompt_description,"input_schema": {"type": "object","properties": {"prompt_name": {"type": "string","description": "Name of the prompt to use. Available: 'generate_issues_prompt'"},"prompt_args": {"type": "object","description": "Arguments for the prompt. For generate_issues_prompt: {opening_brace}'owner': 'repo-owner', 'repo_name': 'repo-name'{closing_brace}","properties": {"owner": {"type": "string","description": "Repository owner (e.g., 'huggingface', 'microsoft')"},"repo_name": {"type": "string","description": "Repository name (e.g., 'transformers', 'vscode')"}}}},"required": ["prompt_name"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelif tool_name == "use_mcp_prompt":# Handle prompt usageprompt_name = tool_args.get("prompt_name")prompt_args = tool_args.get("prompt_args", {})if prompt_name:tool_result = await self.get_prompt(prompt_name, prompt_args)print(f"💭 Prompt '{prompt_name}' generated successfully")result_content = str(tool_result) if tool_result else "Error generating prompt"else:tool_result = "Error: No prompt name provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools, resources, and prompts from the FastMCP server")print()print("💭 PROMPT Examples:")print(" • 'Generate a prompt for asking about issues in facebook/react'")print(" • 'Help me create a good question about microsoft/vscode issues'")print(" • 'I need a structured prompt for analyzing tensorflow/tensorflow'")print()print("🔧 DIRECT Examples:")print(" • 'Show me the issues in huggingface/transformers'")print(" • 'Get repository info for github://repo/google/chrome'")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) < 2 or len(sys.argv) > 3:print("❌ Usage: python client.py <http_server_url> [auth_token]")print("📝 Example: python client.py http://localhost:8000/mcp")print("📝 Example with auth: python client.py http://localhost:8000/mcp <your_bearer_token>")print("📝 Note: Now connects to HTTP server instead of executing script")sys.exit(1)server_url = sys.argv[1]auth_token = sys.argv[2] if len(sys.argv) == 3 else None# Validate URL formatif not server_url.startswith(('http://', 'https://')):print("❌ Error: Server URL must start with http:// or https://")print("📝 Example: python client.py http://localhost:8000")sys.exit(1)# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_url, auth_token)# List available tools, resources, and prompts after connectionawait client.list_available_tools()await client.list_available_resources()await client.list_available_prompts()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Se crea el token de autenticación a partir del token dado por el usuario al iniciar el cliente
# Create authentication if token is provided
auth = None
if auth_token:
auth = BearerAuth(token=auth_token)
print("🔐 Using Bearer token authentication")
else:
print("⚠️ No authentication token provided - connecting without auth")
Se crea el cliente con el token de autenticación, que será enviado al servidor
# Create FastMCP client for HTTP connection using SSE transport
self.client = Client(server_url, auth=auth)
Se conecta con el servidor enviando el token
# Connect to the server
await client.connect_to_server(server_url, auth_token)
Prueba del MCP con autenticación
Como hemos vuelto al http
, primero tenemos que levantar el servidor
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
🔐 Generating RSA key pair for authentication...🎫 Development token generated:eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2dpdGh1Yi1tY3AubWF4Zm4uZGV2Iiwic3ViIjoiZGV2LXVzZXItbWF4Zm4iLCJpYXQiOjE3NTExMDgzMDAsImV4cCI6MTc1MTE5NDcwMCwiYXVkIjoiZ2l0aHViLW1jcC1zZXJ2ZXIiLCJzY29wZSI6ImdpdGh1YjpyZWFkIGdpdGh1Yjp3cml0ZSJ9.PX6BtUhNCv9YVq1ZCh2teAU_LsdGMJx-W2jntTvVgdXv3aDyiOeMuZE9fIcqRy9zcXT1pjexqQQDiRhy8WlRL-mdKooEbIc_ffBVX9LPVaxKAzfzZTnx2lYTt6DgnebjjdNk_OsXF3ujH5s0xmGtY892j-k9P8dJLLrTrqXLhWG2NX_jqHB_kMalFd0LT83D6uXjPako_DKHjYKLc67WvZU_JglVS5eI9YCmmhMlhPHyO4FUlD9xb0DpbOgz8bO1ZExBrB_W2YKomGI_u8R56ItM8bS3eEwybPgEHfHhDNI6PNqsJ3DB1Grmc7KOmGX4LJCfPyB6mpl_bQmChKzcdg💡 Use this token in the client to authenticate------------------------------------------------------------/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:412: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)DEBUG: Starting FastMCP GitHub server...DEBUG: Server name: GitHubMCP[06/28/25 12:58:20] INFO Starting MCP server 'GitHubMCP' with ]8;id=190590;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=102439;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1297\1297]8;;transport 'streamable-http' onhttp://0.0.0.0:8000/mcp/INFO: Started server process [27262]INFO: Waiting for application startup.INFO: Application startup complete.INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Como vemos nos ha generado el token de autentificación eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2dpdGh1Yi1tY3AubWF4Zm4uZGV2Iiwic3ViIjoiZGV2LXVzZXItbWF4Zm4iLCJpYXQiOjE3NTExMDgzMDAsImV4cCI6MTc1MTE5NDcwMCwiYXVkIjoiZ2l0aHViLW1jcC1zZXJ2ZXIiLCJzY29wZSI6ImdpdGh1YjpyZWFkIGdpdGh1Yjp3cml0ZSJ9.PX6BtUhNCv9YVq1ZCh2teAU_LsdGMJx-W2jntTvVgdXv3aDyiOeMuZE9fIcqRy9zcXT1pjexqQQDiRhy8WlRL-mdKooEbIc_ffBVX9LPVaxKAzfzZTnx2lYTt6DgnebjjdNk_OsXF3ujH5s0xmGtY892j-k9P8dJLLrTrqXLhWG2NX_jqHB_kMalFd0LT83D6uXjPako_DKHjYKLc67WvZU_JglVS5eI9YCmmhMlhPHyO4FUlD9xb0DpbOgz8bO1ZExBrB_W2YKomGI_u8R56ItM8bS3eEwybPgEHfHhDNI6PNqsJ3DB1Grmc7KOmGX4LJCfPyB6mpl_bQmChKzcdg
, hay que usarlo a la hora de ejecutar el cliente
Y ahora ejecutamos el cliente con el token de autenticación que nos ha generado el servidor.
!cd client_MCP && source .venv/bin/activate && uv run client.py http://localhost:8000/mcp eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2dpdGh1Yi1tY3AubWF4Zm4uZGV2Iiwic3ViIjoiZGV2LXVzZXItbWF4Zm4iLCJpYXQiOjE3NTExMDgzMDAsImV4cCI6MTc1MTE5NDcwMCwiYXVkIjoiZ2l0aHViLW1jcC1zZXJ2ZXIiLCJzY29wZSI6ImdpdGh1YjpyZWFkIGdpdGh1Yjp3cml0ZSJ9.PX6BtUhNCv9YVq1ZCh2teAU_LsdGMJx-W2jntTvVgdXv3aDyiOeMuZE9fIcqRy9zcXT1pjexqQQDiRhy8WlRL-mdKooEbIc_ffBVX9LPVaxKAzfzZTnx2lYTt6DgnebjjdNk_OsXF3ujH5s0xmGtY892j-k9P8dJLLrTrqXLhWG2NX_jqHB_kMalFd0LT83D6uXjPako_DKHjYKLc67WvZU_JglVS5eI9YCmmhMlhPHyO4FUlD9xb0DpbOgz8bO1ZExBrB_W2YKomGI_u8R56ItM8bS3eEwybPgEHfHhDNI6PNqsJ3DB1Grmc7KOmGX4LJCfPyB6mpl_bQmChKzcdg
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp🔐 Using Bearer token authentication✅ Client created successfully🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain💭 Available prompts (1):==================================================🎯 generate_issues_promptDescription: Generates a structured prompt for asking about GitHub repository issues. Use this when users want to formulate questions about repository issues, or need help creating prompts for issue analysis.Parameters: owner: No description (required), repo_name: No description (required)🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools, resources, and prompts from the FastMCP server💭 PROMPT Examples:• 'Generate a prompt for asking about issues in facebook/react'• 'Help me create a good question about microsoft/vscode issues'• 'I need a structured prompt for analyzing tensorflow/tensorflow'🔧 DIRECT Examples:• 'Show me the issues in huggingface/transformers'• 'Get repository info for github://repo/google/chrome'------------------------------------------------------------👤 You:
Como vemos el cliente se conecta con el servidor y nos da una lista de los tools
, resources
y prompts
disponibles.
Ping del cliente al servidor
Cuando ejecutamos el MCP con http
como capa de transporte, lo normal es que el cliente y el servidor no estén en el mismo ordenador. Por lo que cuando ejecutamos el cliente no podemos saber si el servidor está funcionando, así que podemos desarrollar un ping para comprobar que el servidor está funcionando.
Cliente MCP
Vamos a añadir un ping al cliente MCP
%%writefile client_MCP/client.pyimport sysimport asynciofrom contextlib import AsyncExitStackfrom anthropic import Anthropicfrom dotenv import load_dotenvfrom fastmcp import Clientfrom fastmcp.client.auth import BearerAuth# Load environment variables from .env fileload_dotenv()class FastMCPClient:"""FastMCP client that integrates with Claude to process user queriesand use tools and resources exposed by a FastMCP server."""def __init__(self):"""Initialize the FastMCP client with Anthropic and resource management."""self.exit_stack = AsyncExitStack()self.anthropic = Anthropic()self.client = Noneasync def connect_to_server(self, server_url: str, auth_token: str = None):"""Connect to the specified FastMCP server via HTTP with optional authentication.Args:server_url: URL of the HTTP server (e.g., "http://localhost:8000/mcp")auth_token: Bearer token for authentication (optional)"""print(f"🔗 Connecting to FastMCP HTTP server: {server_url}")# Create authentication if token is providedauth = Noneif auth_token:auth = BearerAuth(token=auth_token)print("🔐 Using Bearer token authentication")else:print("⚠️ No authentication token provided - connecting without auth")# Create FastMCP client for HTTP connection using SSE transportself.client = Client(server_url, auth=auth)# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transportprint("✅ Client created successfully")# Ping to server to check if it's aliveasync with self.client as client:response = await client.ping()print(f"🏓 Server ping response: {response}")async def list_available_tools(self):"""List available tools in the FastMCP server."""try:# Get list of tools from the server using FastMCP contextasync with self.client as client:tools = await client.list_tools()if tools:print(f" 🛠️ Available tools ({len(tools)}):")print("=" * 50)for tool in tools:print(f"📋 {tool.name}")if tool.description:print(f" Description: {tool.description}")# Show parameters if availableif hasattr(tool, 'inputSchema') and tool.inputSchema:if 'properties' in tool.inputSchema:params = list(tool.inputSchema['properties'].keys())print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No tools found in the server")except Exception as e:print(f"❌ Error listing tools: {str(e)}")async def list_available_resources(self):"""List available resources in the FastMCP server."""try:# Get list of resources from the server using FastMCP contextasync with self.client as client:resources = await client.list_resources()if resources:print(f" 📚 Available resources ({len(resources)}):")print("=" * 50)for resource in resources:print(f"📄 {resource.uri}")if resource.name:print(f" Name: {resource.name}")if resource.description:print(f" Description: {resource.description}")if resource.mimeType:print(f" MIME Type: {resource.mimeType}")print()else:print("⚠️ No resources found in the server")except Exception as e:print(f"❌ Error listing resources: {str(e)}")async def list_available_prompts(self):"""List available prompts in the FastMCP server."""try:# Get list of prompts from the server using FastMCP contextasync with self.client as client:prompts = await client.list_prompts()if prompts:print(f" 💭 Available prompts ({len(prompts)}):")print("=" * 50)for prompt in prompts:print(f"🎯 {prompt.name}")if prompt.description:print(f" Description: {prompt.description}")# Show parameters if availableif hasattr(prompt, 'arguments') and prompt.arguments:params = []for arg in prompt.arguments:param_info = f"{arg.name}: {arg.description or 'No description'}"if arg.required:param_info += " (required)"params.append(param_info)print(f" Parameters: {', '.join(params)}")print()else:print("⚠️ No prompts found in the server")except Exception as e:print(f"❌ Error listing prompts: {str(e)}")async def read_resource(self, resource_uri: str):"""Read a specific resource from the server.Args:resource_uri: URI of the resource to readReturns:str: Resource content"""try:async with self.client as client:result = await client.read_resource(resource_uri)return resultexcept Exception as e:print(f"❌ Error reading resource {resource_uri}: {str(e)}")return Noneasync def get_prompt(self, prompt_name: str, prompt_args: dict = None):"""Get/call a specific prompt from the server.Args:prompt_name: Name of the prompt to callprompt_args: Arguments for the prompt (if any)Returns:str: Generated prompt content"""try:async with self.client as client:if prompt_args:result = await client.get_prompt(prompt_name, prompt_args)else:result = await client.get_prompt(prompt_name)# Extract the prompt text from the responseif hasattr(result, 'messages') and result.messages:# FastMCP returns prompts as message objectsreturn ' '.join([msg.content.text for msg in result.messages if hasattr(msg.content, 'text')])elif hasattr(result, 'content'):return str(result.content)else:return str(result)except Exception as e:print(f"❌ Error getting prompt {prompt_name}: {str(e)}")return Noneasync def process_query(self, query: str) -> str:"""Process a user query, interacting with Claude and FastMCP tools and resources.Args:query: User queryReturns:str: Final processed response"""try:# Use FastMCP context for all operationsasync with self.client as client:# Get available tools and resourcestools_list = await client.list_tools()resources_list = await client.list_resources()# Prepare tools for Claude in correct formatclaude_tools = []for tool in tools_list:claude_tool = {"name": tool.name,"description": tool.description or f"Tool {tool.name}","input_schema": tool.inputSchema or {"type": "object", "properties": {}}}claude_tools.append(claude_tool)# Add a special tool for reading resources (including template resources)resource_description = "Read a resource from the MCP server. "if resources_list:# Convert URIs to strings to avoid AnyUrl object issuesresource_uris = [str(r.uri) for r in resources_list]resource_description += f"Available static resources: {', '.join(resource_uris)}. "resource_description += "Also supports template resources like github://repo/owner/repo_name for GitHub repository information."claude_tools.append({"name": "read_mcp_resource","description": resource_description,"input_schema": {"type": "object","properties": {"resource_uri": {"type": "string","description": "URI of the resource to read. Can be static (like resource://server_info) or template-based (like github://repo/facebook/react)"}},"required": ["resource_uri"]}})# Add a special tool for using promptsprompt_description = "Generate specialized prompts from the MCP server. Use this when users want to: "prompt_description += "- Create well-structured questions about repositories "prompt_description += "- Get help formulating prompts for specific tasks "prompt_description += "- Generate template questions for analysis "if prompts_list:prompt_names = [p.name for p in prompts_list]prompt_description += f" Available prompts: {', '.join(prompt_names)} "prompt_description += "- generate_issues_prompt: Creates structured questions about GitHub repository issues"prompt_description += " IMPORTANT: Use prompts when users explicitly ask for help creating questions or prompts, or when they want to formulate better questions about repositories."claude_tools.append({"name": "use_mcp_prompt","description": prompt_description,"input_schema": {"type": "object","properties": {"prompt_name": {"type": "string","description": "Name of the prompt to use. Available: 'generate_issues_prompt'"},"prompt_args": {"type": "object","description": "Arguments for the prompt. For generate_issues_prompt: {opening_brace}'owner': 'repo-owner', 'repo_name': 'repo-name'{closing_brace}","properties": {"owner": {"type": "string","description": "Repository owner (e.g., 'huggingface', 'microsoft')"},"repo_name": {"type": "string","description": "Repository name (e.g., 'transformers', 'vscode')"}}}},"required": ["prompt_name"]}})# Create initial message for Claudemessages = [{"role": "user","content": query}]# First call to Clauderesponse = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Process Claude's responseresponse_text = ""for content_block in response.content:if content_block.type == "text":response_text += content_block.textelif content_block.type == "tool_use":# Claude wants to use a tooltool_name = content_block.nametool_args = content_block.inputtool_call_id = content_block.idprint(f"🔧 Claude wants to use: {tool_name}")print(f"📝 Arguments: {tool_args}")try:if tool_name == "read_mcp_resource":# Handle resource readingresource_uri = tool_args.get("resource_uri")if resource_uri:tool_result = await client.read_resource(resource_uri)print(f"📖 Resource read successfully: {resource_uri}")# Better handling of resource resultif hasattr(tool_result, 'content'):# If it's a resource response object, extract contentif hasattr(tool_result.content, 'text'):result_content = tool_result.content.textelse:result_content = str(tool_result.content)else:# If it's already a string or simple objectresult_content = str(tool_result)else:tool_result = "Error: No resource URI provided"result_content = tool_resultelif tool_name == "use_mcp_prompt":# Handle prompt usageprompt_name = tool_args.get("prompt_name")prompt_args = tool_args.get("prompt_args", {})if prompt_name:tool_result = await self.get_prompt(prompt_name, prompt_args)print(f"💭 Prompt '{prompt_name}' generated successfully")result_content = str(tool_result) if tool_result else "Error generating prompt"else:tool_result = "Error: No prompt name provided"result_content = tool_resultelse:# Execute regular tool on the FastMCP servertool_result = await client.call_tool(tool_name, tool_args)print(f"✅ Tool executed successfully")result_content = str(tool_result)# Add tool result to the conversationmessages.append({"role": "assistant","content": response.content})# Format result for Claudeif tool_result:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": f"Tool result: {result_content}"}]})else:messages.append({"role": "user","content": [{"type": "tool_result","tool_use_id": tool_call_id,"content": "Tool executed without response content"}]})# Second call to Claude with the tool resultfinal_response = self.anthropic.messages.create(model="claude-3-5-sonnet-20241022",max_tokens=6000,messages=messages,tools=claude_tools if claude_tools else None)# Extract text from the final responsefor final_content in final_response.content:if final_content.type == "text":response_text += final_content.textexcept Exception as e:error_msg = f"❌ Error executing {tool_name}: {str(e)}"print(error_msg)response_text += f" {error_msg}"return response_textexcept Exception as e:error_msg = f"❌ Error processing query: {str(e)}"print(error_msg)return error_msgasync def chat_loop(self):"""Main chat loop with user interaction."""print(" 🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.")print("💬 You can ask questions about GitHub repositories!")print("📚 The client can use tools, resources, and prompts from the FastMCP server")print()print("💭 PROMPT Examples:")print(" • 'Generate a prompt for asking about issues in facebook/react'")print(" • 'Help me create a good question about microsoft/vscode issues'")print(" • 'I need a structured prompt for analyzing tensorflow/tensorflow'")print()print("🔧 DIRECT Examples:")print(" • 'Show me the issues in huggingface/transformers'")print(" • 'Get repository info for github://repo/google/chrome'")print("-" * 60)while True:try:# Request user inputuser_input = input(" 👤 You: ").strip()if user_input.lower() in ['quit', 'q', 'exit', 'salir']:print("👋 Bye!")breakif not user_input:continueprint(" 🤔 Claude is thinking...")# Process queryresponse = await self.process_query(user_input)# Show responseprint(f" 🤖 Claude: {response}")except KeyboardInterrupt:print(" 👋 Disconnecting...")breakexcept Exception as e:print(f" ❌ Error in chat: {str(e)}")continueasync def cleanup(self):"""Clean up resources and close connections."""print("🧹 Cleaning up resources...")# FastMCP Client cleanup is handled automatically by context managerawait self.exit_stack.aclose()print("✅ Resources released")async def main():"""Main function that initializes and runs the FastMCP client."""# Verify command line argumentsif len(sys.argv) < 2 or len(sys.argv) > 3:print("❌ Usage: python client.py <http_server_url> [auth_token]")print("📝 Example: python client.py http://localhost:8000/mcp")print("📝 Example with auth: python client.py http://localhost:8000/mcp <your_bearer_token>")print("📝 Note: Now connects to HTTP server instead of executing script")sys.exit(1)server_url = sys.argv[1]auth_token = sys.argv[2] if len(sys.argv) == 3 else None# Validate URL formatif not server_url.startswith(('http://', 'https://')):print("❌ Error: Server URL must start with http:// or https://")print("📝 Example: python client.py http://localhost:8000")sys.exit(1)# Create and run clientclient = FastMCPClient()try:# Connect to the serverawait client.connect_to_server(server_url, auth_token)# List available tools, resources, and prompts after connectionawait client.list_available_tools()await client.list_available_resources()# Start chat loopawait client.chat_loop()except Exception as e:print(f"❌ Fatal error: {str(e)}")finally:# Ensure resources are cleaned upawait client.cleanup()if __name__ == "__main__":# Entry point of the scriptasyncio.run(main())
Overwriting client_MCP/client.py
Añadimos en el método connect_to_server
el ping
# Ping to server to check if it's alive
async with self.client as client:
response = await client.ping()
print(f"🏓 Server ping response: {response}")
Prueba del ping
Levantamos primero el servidor
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
🔐 Generating RSA key pair for authentication...🎫 Development token generated:eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2dpdGh1Yi1tY3AubWF4Zm4uZGV2Iiwic3ViIjoiZGV2LXVzZXItbWF4Zm4iLCJpYXQiOjE3NTExMDkxMTIsImV4cCI6MTc1MTE5NTUxMiwiYXVkIjoiZ2l0aHViLW1jcC1zZXJ2ZXIiLCJzY29wZSI6ImdpdGh1YjpyZWFkIGdpdGh1Yjp3cml0ZSJ9.N_3QPIHW3BSn1iSSkrcaoelbwA-0D9Z3gelILb8fu1JC2JhCgtnJ0IwNqJrVhAkU0CNcykT36Q3mpCgy0hDhnFKkO9SRGVFgSw71voF5YNOkzzBY14cJERolYy9UDZA6geHxwR0rKyCGYkDH-NAKPuYWC9K7UlGfuOuzh3mp-XQ3Zy4mkyvfhiuwuaJ5_MdR0YtJj6opSRbEsVs1PtFYZETPExx3iBGck2qzLek-LxAJ6mjagPncikWeDwaYShFNPO0Ub3wm2Ok_ak_TChmN3W15MknfBXZrKcIhsNIhCrXJjZkSezp5JX49zoljdK2By9-QH1xmWCQqif_APD-hNQ💡 Use this token in the client to authenticate------------------------------------------------------------/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/github_server.py:412: DeprecationWarning: Mount prefixes are now optional and the first positional argument should be the server you want to mount.mcp.mount("sub_mcp", sub_mcp)DEBUG: Starting FastMCP GitHub server...DEBUG: Server name: GitHubMCP[06/28/25 13:11:52] INFO Starting MCP server 'GitHubMCP' with ]8;id=186381;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.pyserver.py]8;;:]8;id=502881;file:///Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server/.venv/lib/python3.11/site-packages/fastmcp/server/server.py#1297\1297]8;;transport 'streamable-http' onhttp://0.0.0.0:8000/mcp/INFO: Started server process [31017]INFO: Waiting for application startup.INFO: Application startup complete.INFO: Uvicorn running on http://0.0.0.0:8000 (Press CTRL+C to quit)
Y ahora ejecutamos el cliente con el token de autenticación
!cd client_MCP && source .venv/bin/activate && uv run client.py http://localhost:8000/mcp eyJhbGciOiJSUzI1NiIsInR5cCI6IkpXVCJ9.eyJpc3MiOiJodHRwczovL2dpdGh1Yi1tY3AubWF4Zm4uZGV2Iiwic3ViIjoiZGV2LXVzZXItbWF4Zm4iLCJpYXQiOjE3NTExMDkxMTIsImV4cCI6MTc1MTE5NTUxMiwiYXVkIjoiZ2l0aHViLW1jcC1zZXJ2ZXIiLCJzY29wZSI6ImdpdGh1YjpyZWFkIGdpdGh1Yjp3cml0ZSJ9.N_3QPIHW3BSn1iSSkrcaoelbwA-0D9Z3gelILb8fu1JC2JhCgtnJ0IwNqJrVhAkU0CNcykT36Q3mpCgy0hDhnFKkO9SRGVFgSw71voF5YNOkzzBY14cJERolYy9UDZA6geHxwR0rKyCGYkDH-NAKPuYWC9K7UlGfuOuzh3mp-XQ3Zy4mkyvfhiuwuaJ5_MdR0YtJj6opSRbEsVs1PtFYZETPExx3iBGck2qzLek-LxAJ6mjagPncikWeDwaYShFNPO0Ub3wm2Ok_ak_TChmN3W15MknfBXZrKcIhsNIhCrXJjZkSezp5JX49zoljdK2By9-QH1xmWCQqif_APD-hNQ
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp🔐 Using Bearer token authentication✅ Client created successfully🏓 Server ping response: True🛠️ Available tools (2):==================================================📋 sub_mcp_hello_worldDescription: Returns a simple greeting.Parameters:📋 list_repository_issuesDescription: Lists open issues for a given GitHub repository.Args:owner: The owner of the repository (e.g., 'modelcontextprotocol')repo_name: The name of the repository (e.g., 'python-sdk')ctx: The context of the requestuser_id: The user ID (automatically injected by the server)Returns:list[dict]: A list of dictionaries, each containing information about an issueParameters: owner, repo_name📚 Available resources (1):==================================================📄 resource://server_infoName: server_infoDescription: Returns information about the server.MIME Type: text/plain🤖 FastMCP HTTP client started. Write 'quit', 'q', 'exit', 'salir' to exit.💬 You can ask questions about GitHub repositories!📚 The client can use tools, resources, and prompts from the FastMCP server💭 PROMPT Examples:• 'Generate a prompt for asking about issues in facebook/react'• 'Help me create a good question about microsoft/vscode issues'• 'I need a structured prompt for analyzing tensorflow/tensorflow'🔧 DIRECT Examples:• 'Show me the issues in huggingface/transformers'• 'Get repository info for github://repo/google/chrome'------------------------------------------------------------👤 You: q👋 Bye!🧹 Cleaning up resources...✅ Resources released
Como vemos el servidor ha respondido al ping
🏓 Server ping response: True