MCP: Guía Completa para Crear servidores y clientes MCP (Model Context Protocol) con FastMCP

MCP: Guía Completa para Crear servidores y clientes MCP (Model Context Protocol) con FastMCP MCP: Guía Completa para Crear servidores y clientes MCP (Model Context Protocol) con FastMCP

¿Qué es MCP?link image 1

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.

MCP vs API

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.

USB MCP

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

mcp architecture

FastMCPlink image 2

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 virtuallink image 3

Para crear un servidor y un cliente MCP, vamos a crear entornos virtuales con uv con las dependencias que vamos a necesitar

Servidor MCPlink image 4

Primero creamos una carpeta para el servidor de MCP

	
!mkdir gitHub_MCP_server
Copy

Iniciamos el entorno uv

	
!cd gitHub_MCP_server && uv init .
Copy
	
Initialized project `github-mcp-server` at `/Users/macm1/Documents/web/portafolio/posts/gitHub_MCP_server`

Lo activamos

	
!cd gitHub_MCP_server && uv venv
Copy
	
Using CPython 3.11.11
Creating virtual environment at: .venv
Activate with: source .venv/bin/activate

E instalamos las librerías necesarias

	
!cd gitHub_MCP_server && uv add anthropic fastmcp python-dotenv requests
Copy
	
Resolved 42 packages in 34ms
Installed 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 MCPlink image 5

Ahora creamos una carpeta donde programaremos el cliente MCP

	
!mkdir client_MCP
Copy

Iniciamos el entorno uv

	
!cd client_MCP && uv init .
Copy
	
Initialized project `client-mcp` at `/Users/macm1/Documents/web/portafolio/posts/client_MCP`

Lo activamos

	
!cd client_MCP && uv venv
Copy
	
Using CPython 3.11.11
Creating virtual environment at: .venv
Activate 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
Copy
	
Resolved 42 packages in 307ms
Prepared 5 packages in 115ms
Installed 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/.env
ANTHROPIC_API_KEY="ANTHROPIC_API_KEY"
Copy
	
Writing client_MCP/.env

MCP básicolink image 6

Escribimos el mínimo código que necesitamos para tener un servidor MCP

	
%%writefile gitHub_MCP_server/github_server.py
from mcp.server.fastmcp import FastMCP
# Create an MCP server
mcp = FastMCP("GitHubMCP")
if __name__ == "__main__":
# Initialize and run the server
mcp.run(transport='stdio')
Copy
	
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 GitHublink image 7

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.py
import os
from dotenv import load_dotenv
# Load the GitHub token from the .env file
load_dotenv()
GITHUB_TOKEN = os.environ.get("GITHUB_TOKEN")
# Check if the GitHub token is configured
if 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 requests
def create_github_headers():
headers = {}
if GITHUB_TOKEN:
headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
# GitHub recommends including a User-Agent
headers["User-Agent"] = "MCP_GitHub_Server_Example"
headers["Accept"] = "application/vnd.github.v3+json" # Good practice
return headers
Copy
	
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/.env
GITHUB_TOKEN = "GITHUB_TOKEN"
Copy
	
Overwriting gitHub_MCP_server/.env

Crear tool de MCP para obtener una lista de issues de un repositorio de GitHub

Servidor MCPlink image 8

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.py
import httpx
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create a FastMCP server
mcp = 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 responses
api_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 summary
summary = 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 information
result = {
"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 server
mcp.run()
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Cliente MCPlink image 9

Ahora creamos un cliente MCP para poder usar la tool que hemos creado

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 extension
if not server_script_path.endswith('.py'):
raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")
# Create FastMCP client
self.client = Client(server_script_path)
# Note: FastMCP Client automatically infers transport from .py files
print("✅ 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 context
async 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 available
if 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 query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools
tools_list = await client.list_tools()
# Prepare tools for Claude in correct format
claude_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
# Execute tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if tool_result:
# Convert result to string format for Claude
result_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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if 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 client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_script_path)
# List available tools after connection
await client.list_available_tools()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
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 tools disponibles con el método list_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 introduce quit, q, exit o salir en el chat.
  • Se procesa la entrada del usuario con el método process_query que obtiene la lista de tools disponibles y hace una petición al LLM con el mensaje del usuario y la lista de tools
    • 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 la tool, 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
Copy
	
🔗 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_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
🤖 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 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?
👤 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ónlink image 10

Servidor MCPlink image 11

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.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
mcp = 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 responses
api_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 summary
summary = 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 information
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
}
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 server
mcp.run()
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Filtrar tools mediante tags

Servidor MCPlink image 12

MCP nos da la opción de poder exponer tools mediante tags, lo cual puede ser útil para exponer solo tools 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.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
mcp = 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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 server
mcp.run()
Copy
	
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 tools que tengan el tag public, es decir, a list_repository_issues. Solo va a poder ver una lista de 10 issues.

Prueba de los tagslink image 13

Volvemos a ejecutar el cliente MCP

	
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
Copy
	
🔗 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_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
🤖 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 privatelink image 14

Cambiamos include_tags a private para usar la tool list_more_repository_issues

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
mcp = 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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 server
mcp.run()
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Prueba del tag privatelink image 15

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
Copy
	
🔗 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_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
🤖 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 tools 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 publiclink image 16

Volvemos a poner include_tags a public para usar la tool list_repository_issues

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
mcp = 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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 server
mcp.run()
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Excluir tools por tags

Al igual que antes hemos filtrado las tools que se pueden usar por tags, también podemos excluir tools 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 MCPlink image 17

Creamos una nueva tool y la excluimos mediante tags

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
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"}
)
@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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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 server
mcp.run()
Copy

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
Copy
	
🔗 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_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
🤖 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 servidoreslink image 18

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 MCPlink image 19

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.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
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"}
)
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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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
mcp.run()
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Prueba de la composición de servidores MCPlink image 20

Ejecutamos el cliente

	
!cd client_MCP && source .venv/bin/activate && uv run client.py ../gitHub_MCP_server/github_server.py
Copy
	
🔗 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_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
🤖 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 transportelink image 21

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 MCPlink image 22

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.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
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"}
)
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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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/mcp
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
)
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Cliente MCPlink image 23

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.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 transport
self.client = Client(server_url)
# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transport
print("✅ 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 context
async 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 available
if 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 query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools
tools_list = await client.list_tools()
# Prepare tools for Claude in correct format
claude_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
# Execute tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if tool_result:
# Convert result to string format for Claude
result_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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if 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 format
if 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 client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_url)
# List available tools after connection
await client.list_available_tools()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
Overwriting client_MCP/client.py

Prueba del MCP por httplink image 24

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
Copy
	
/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' on
http://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
Copy
	
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp
✅ Client created successfully
🛠️ 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
🤖 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.py
import httpx
from typing import Optional
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
# Create FastMCP server
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"}
)
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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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
mcp.run(
transport="stdio"
)
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Argumentos excluídoslink image 25

Servidor MCPlink image 26

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.py
import httpx
from fastmcp import FastMCP
from github import GITHUB_TOKEN, create_github_headers
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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
mcp.run(
transport="stdio"
)
Copy
	
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.

Contextolink image 27

Podemos pasar información de contexto del servidor al cliente y viceversa.

Servidor MCPlink image 28

Vamos a añadir contexto a nuestro servidor MCP.

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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 request
user_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 responses
api_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 summary
summary = 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 information
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
}
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 responses
api_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 summary
summary = 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 information
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
}
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 order
api_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 issue
summary = 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 information
result = {
"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
mcp.run(
transport="stdio"
)
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Hemos sustituído todos los prints 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 MCPlink image 29

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.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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() -&gt; 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() -&gt; 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
mcp.run(
transport="stdio"
)
Copy
	
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 resources

Además, le hemos puesto la tag public, ya que nuestro servidor MCP solo incluye las tools o resources 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 MCPlink image 30

Ahora tenemos que hacer que nuestro cliente pueda ver los resources de nuestro servidor MCP.

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 extension
if not server_script_path.endswith('.py'):
raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")
# Create FastMCP client
self.client = Client(server_script_path)
# Note: FastMCP Client automatically infers transport from .py files
print("✅ 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 context
async 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 available
if 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 context
async 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools and resources
tools_list = await client.list_tools()
resources_list = await client.list_resources()
# Prepare tools for Claude in correct format
claude_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
if resources_list:
# Convert URIs to strings to avoid AnyUrl object issues
resource_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) != 2:
print("❌ Usage: python client.py &lt;path_to_fastmcp_server&gt;")
print("📝 Example: python client.py ../MCP_github/github_server.py")
sys.exit(1)
server_script_path = sys.argv[1]
# Create and run client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_script_path)
# List available tools and resources after connection
await client.list_available_tools()
await client.list_available_resources()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
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
Copy
	
🔗 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_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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: 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 resources

📚 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 tools, podemos añadir contexto a los resources.

Servidor MCPlink image 31

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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) -&gt; 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() -&gt; 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
mcp.run(
transport="stdio"
)
Copy
	
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
Copy
	
🔗 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_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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: 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 templates, 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 MCPlink image 32

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.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
import datetime
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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) -&gt; 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) -&gt; 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 request
Returns:
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 information
repo_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_info
except 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() -&gt; 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
mcp.run(
transport="stdio"
)
Copy
	
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 MCPlink image 33

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.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 extension
if not server_script_path.endswith('.py'):
raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")
# Create FastMCP client
self.client = Client(server_script_path)
# Note: FastMCP Client automatically infers transport from .py files
print("✅ 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 context
async 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 available
if 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 context
async 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools and resources
tools_list = await client.list_tools()
resources_list = await client.list_resources()
# Prepare tools for Claude in correct format
claude_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 issues
resource_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) != 2:
print("❌ Usage: python client.py &lt;path_to_fastmcp_server&gt;")
print("📝 Example: python client.py ../MCP_github/github_server.py")
sys.exit(1)
server_script_path = sys.argv[1]
# Create and run client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_script_path)
# List available tools and resources after connection
await client.list_available_tools()
await client.list_available_resources()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
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
Copy
	
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
🛠️ 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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: 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 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.
👤 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 MCPlink image 34

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 tools, los resources y los prompts 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.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
import datetime
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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) -&gt; 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) -&gt; 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 request
Returns:
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 information
repo_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_info
except 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) -&gt; str:
"""
Generates a structured prompt for asking about GitHub repository issues.
This prompt template helps users formulate clear questions about repository issues
and 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 issues
Repository: {owner}/{repo_name}"""
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; 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
mcp.run(
transport="stdio"
)
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Cliente MCPlink image 35

Modificamos nuestro cliente para poder usar el prompt que hemos creado en nuestro servidor.

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 extension
if not server_script_path.endswith('.py'):
raise ValueError(f"Unsupported server type. Use .py files. Received: {server_script_path}")
# Create FastMCP client
self.client = Client(server_script_path)
# Note: FastMCP Client automatically infers transport from .py files
print("✅ 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 context
async 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 available
if 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 context
async 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 context
async 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 available
if 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async 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 call
prompt_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 response
if hasattr(result, 'messages') and result.messages:
# FastMCP returns prompts as message objects
return ' '.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 None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools, resources, and prompts
tools_list = await client.list_tools()
resources_list = await client.list_resources()
prompts_list = await client.list_prompts()
# Prepare tools for Claude in correct format
claude_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 issues
resource_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 prompts
prompt_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
elif tool_name == "use_mcp_prompt":
# Handle prompt usage
prompt_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_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) != 2:
print("❌ Usage: python client.py &lt;path_to_fastmcp_server&gt;")
print("📝 Example: python client.py ../MCP_github/github_server.py")
sys.exit(1)
server_script_path = sys.argv[1]
# Create and run client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_script_path)
# List available tools, resources, and prompts after connection
await client.list_available_tools()
await client.list_available_resources()
await client.list_available_prompts()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
Overwriting client_MCP/client.py

Hemos creado las funciones list_available_prompts y get_prompt para listar los prompts 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
Copy
	
🔗 Connecting to FastMCP server: ../gitHub_MCP_server/github_server.py
✅ Client created successfully
🛠️ 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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: Returns information about the server.
MIME Type: text/plain
💭 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'
------------------------------------------------------------
👤 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 prompts 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 HTTPlink image 36

Volvemos a configurar http como capa de transporte para las dos últimas cosas que vamos a ver

Servidor MCPlink image 37

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from fastmcp import FastMCP, Context
from github import GITHUB_TOKEN, create_github_headers
import datetime
USER_ID = 1234567890
# Create FastMCP server
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"}
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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) -&gt; 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) -&gt; 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 request
Returns:
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 information
repo_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_info
except 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) -&gt; str:
"""
Generates a structured prompt for asking about GitHub repository issues.
This prompt template helps users formulate clear questions about repository issues
and 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 issues
Repository: {owner}/{repo_name}"""
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; 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
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
)
Copy
	
Overwriting gitHub_MCP_server/github_server.py

Cliente MCPlink image 38

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 transport
self.client = Client(server_url)
# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transport
print("✅ 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 context
async 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 available
if 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 context
async 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools and resources
tools_list = await client.list_tools()
resources_list = await client.list_resources()
# Prepare tools for Claude in correct format
claude_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 issues
resource_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) != 2:
print("❌ Usage: python client.py &lt;http_server_url&gt;")
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 format
if 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 client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_url)
# List available tools and resources after connection
await client.list_available_tools()
await client.list_available_resources()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
Overwriting client_MCP/client.py

Autenticaciónlink image 39

Si queremos crear un servidor MCP al que solo se pueda conectar determinados clientes, podemos añadir autenticación

Servidor MCPlink image 40

Creamos el servidor con autenticación

	
%%writefile gitHub_MCP_server/github_server.py
import httpx
from fastmcp import FastMCP, Context
from fastmcp.server.auth import BearerAuthProvider
from fastmcp.server.auth.providers.bearer import RSAKeyPair
from fastmcp.server.dependencies import get_access_token, AccessToken
from github import GITHUB_TOKEN, create_github_headers
import datetime
USER_ID = 1234567890
# 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)
# Create FastMCP server with authentication
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
)
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) -&gt; 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 request
user_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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
try:
access_token: AccessToken = get_access_token()
authenticated_user = access_token.client_id
user_scopes = access_token.scopes
ctx.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 information
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,
"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) -&gt; 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 responses
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=open&amp;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 summary
summary = f"#{issue.get('number', 'N/A')}: {issue.get('title', 'No title')}"
if issue.get('comments', 0) &gt; 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 information
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
}
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) -&gt; 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 order
api_url = f"https://api.github.com/repos/{owner}/{repo_name}/issues?state=all&amp;sort=created&amp;direction=asc&amp;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 issue
summary = f"#{first_issue.get('number', 'N/A')}: {first_issue.get('title', 'No title')}"
if first_issue.get('comments', 0) &gt; 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", "")) &gt; 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 information
result = {
"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) -&gt; 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) -&gt; 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 request
Returns:
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 information
repo_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_info
except 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) -&gt; str:
"""
Generates a structured prompt for asking about GitHub repository issues.
This prompt template helps users formulate clear questions about repository issues
and 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 issues
Repository: {owner}/{repo_name}"""
@sub_mcp.tool(tags={"public"})
def hello_world() -&gt; 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 &lt;your_bearer_token&gt;
mcp.run(
transport="streamable-http",
host="0.0.0.0",
port=8000,
)
Copy
	
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 MCPlink image 41

Creamos el cliente MCP con autenticación

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
from fastmcp.client.auth import BearerAuth
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 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")
# Create FastMCP client for HTTP connection using SSE transport
self.client = Client(server_url, auth=auth)
# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transport
print("✅ 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 context
async 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 available
if 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 context
async 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 context
async 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 available
if 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async 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 call
prompt_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 response
if hasattr(result, 'messages') and result.messages:
# FastMCP returns prompts as message objects
return ' '.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 None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools and resources
tools_list = await client.list_tools()
resources_list = await client.list_resources()
# Prepare tools for Claude in correct format
claude_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 issues
resource_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 prompts
prompt_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
elif tool_name == "use_mcp_prompt":
# Handle prompt usage
prompt_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_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) &lt; 2 or len(sys.argv) &gt; 3:
print("❌ Usage: python client.py &lt;http_server_url&gt; [auth_token]")
print("📝 Example: python client.py http://localhost:8000/mcp")
print("📝 Example with auth: python client.py http://localhost:8000/mcp &lt;your_bearer_token&gt;")
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 format
if 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 client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_url, auth_token)
# List available tools, resources, and prompts after connection
await client.list_available_tools()
await client.list_available_resources()
await client.list_available_prompts()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
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ónlink image 42

Como hemos vuelto al http, primero tenemos que levantar el servidor

	
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
Copy
	
🔐 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' on
http://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
Copy
	
🔗 Connecting to FastMCP HTTP server: http://localhost:8000/mcp
🔐 Using Bearer token authentication
✅ Client created successfully
🛠️ 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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: Returns information about the server.
MIME Type: text/plain
💭 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 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 servidorlink image 43

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 MCPlink image 44

Vamos a añadir un ping al cliente MCP

	
%%writefile client_MCP/client.py
import sys
import asyncio
from contextlib import AsyncExitStack
from anthropic import Anthropic
from dotenv import load_dotenv
from fastmcp import Client
from fastmcp.client.auth import BearerAuth
# Load environment variables from .env file
load_dotenv()
class FastMCPClient:
"""
FastMCP client that integrates with Claude to process user queries
and 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 = None
async 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 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")
# Create FastMCP client for HTTP connection using SSE transport
self.client = Client(server_url, auth=auth)
# Note: FastMCP Client automatically detects HTTP URLs and uses SSE transport
print("✅ Client created successfully")
# 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}")
async def list_available_tools(self):
"""List available tools in the FastMCP server."""
try:
# Get list of tools from the server using FastMCP context
async 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 available
if 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 context
async 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 context
async 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 available
if 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 read
Returns:
str: Resource content
"""
try:
async with self.client as client:
result = await client.read_resource(resource_uri)
return result
except Exception as e:
print(f"❌ Error reading resource {resource_uri}: {str(e)}")
return None
async 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 call
prompt_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 response
if hasattr(result, 'messages') and result.messages:
# FastMCP returns prompts as message objects
return ' '.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 None
async def process_query(self, query: str) -&gt; str:
"""
Process a user query, interacting with Claude and FastMCP tools and resources.
Args:
query: User query
Returns:
str: Final processed response
"""
try:
# Use FastMCP context for all operations
async with self.client as client:
# Get available tools and resources
tools_list = await client.list_tools()
resources_list = await client.list_resources()
# Prepare tools for Claude in correct format
claude_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 issues
resource_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 prompts
prompt_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 Claude
messages = [
{
"role": "user",
"content": query
}
]
# First call to Claude
response = 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 response
response_text = ""
for content_block in response.content:
if content_block.type == "text":
response_text += content_block.text
elif content_block.type == "tool_use":
# Claude wants to use a tool
tool_name = content_block.name
tool_args = content_block.input
tool_call_id = content_block.id
print(f"🔧 Claude wants to use: {tool_name}")
print(f"📝 Arguments: {tool_args}")
try:
if tool_name == "read_mcp_resource":
# Handle resource reading
resource_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 result
if hasattr(tool_result, 'content'):
# If it's a resource response object, extract content
if hasattr(tool_result.content, 'text'):
result_content = tool_result.content.text
else:
result_content = str(tool_result.content)
else:
# If it's already a string or simple object
result_content = str(tool_result)
else:
tool_result = "Error: No resource URI provided"
result_content = tool_result
elif tool_name == "use_mcp_prompt":
# Handle prompt usage
prompt_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_result
else:
# Execute regular tool on the FastMCP server
tool_result = await client.call_tool(tool_name, tool_args)
print(f"✅ Tool executed successfully")
result_content = str(tool_result)
# Add tool result to the conversation
messages.append({
"role": "assistant",
"content": response.content
})
# Format result for Claude
if 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 result
final_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 response
for final_content in final_response.content:
if final_content.type == "text":
response_text += final_content.text
except Exception as e:
error_msg = f"❌ Error executing {tool_name}: {str(e)}"
print(error_msg)
response_text += f" {error_msg}"
return response_text
except Exception as e:
error_msg = f"❌ Error processing query: {str(e)}"
print(error_msg)
return error_msg
async 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 input
user_input = input(" 👤 You: ").strip()
if user_input.lower() in ['quit', 'q', 'exit', 'salir']:
print("👋 Bye!")
break
if not user_input:
continue
print(" 🤔 Claude is thinking...")
# Process query
response = await self.process_query(user_input)
# Show response
print(f" 🤖 Claude: {response}")
except KeyboardInterrupt:
print(" 👋 Disconnecting...")
break
except Exception as e:
print(f" ❌ Error in chat: {str(e)}")
continue
async def cleanup(self):
"""Clean up resources and close connections."""
print("🧹 Cleaning up resources...")
# FastMCP Client cleanup is handled automatically by context manager
await self.exit_stack.aclose()
print("✅ Resources released")
async def main():
"""
Main function that initializes and runs the FastMCP client.
"""
# Verify command line arguments
if len(sys.argv) &lt; 2 or len(sys.argv) &gt; 3:
print("❌ Usage: python client.py &lt;http_server_url&gt; [auth_token]")
print("📝 Example: python client.py http://localhost:8000/mcp")
print("📝 Example with auth: python client.py http://localhost:8000/mcp &lt;your_bearer_token&gt;")
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 format
if 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 client
client = FastMCPClient()
try:
# Connect to the server
await client.connect_to_server(server_url, auth_token)
# List available tools, resources, and prompts after connection
await client.list_available_tools()
await client.list_available_resources()
# Start chat loop
await client.chat_loop()
except Exception as e:
print(f"❌ Fatal error: {str(e)}")
finally:
# Ensure resources are cleaned up
await client.cleanup()
if __name__ == "__main__":
# Entry point of the script
asyncio.run(main())
Copy
	
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 pinglink image 45

Levantamos primero el servidor

	
!cd gitHub_MCP_server && source .venv/bin/activate && uv run github_server.py
Copy
	
🔐 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' on
http://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
Copy
	
🔗 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_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')
ctx: The context of the request
user_id: The user ID (automatically injected by the server)
Returns:
list[dict]: A list of dictionaries, each containing information about an issue
Parameters: owner, repo_name
📚 Available resources (1):
==================================================
📄 resource://server_info
Name: server_info
Description: 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

Seguir leyendo

Últimos posts -->

¿Has visto estos proyectos?

Horeca chatbot

Horeca chatbot Horeca chatbot
Python
LangChain
PostgreSQL
PGVector
React
Kubernetes
Docker
GitHub Actions

Chatbot conversacional para cocineros de hoteles y restaurantes. Un cocinero, jefe de cocina o camaeror de un hotel o restaurante puede hablar con el chatbot para obtener información de recetas y menús. Pero además implementa agentes, con los cuales puede editar o crear nuevas recetas o menús

Naviground

Naviground Naviground

Subtify

Subtify Subtify
Python
Whisper
Spaces

Generador de subtítulos para videos en el idioma que desees. Además a cada persona le pone su subtítulo de un color

Ver todos los proyectos -->

¿Quieres aplicar la IA en tu proyecto? Contactame!

¿Quieres mejorar con estos tips?

Últimos tips -->

Usa esto en local

Los espacios de Hugging Face nos permite ejecutar modelos con demos muy sencillas, pero ¿qué pasa si la demo se rompe? O si el usuario la elimina? Por ello he creado contenedores docker con algunos espacios interesantes, para poder usarlos de manera local, pase lo que pase. De hecho, es posible que si pinchas en alún botón de ver proyecto te lleve a un espacio que no funciona.

Flow edit

Flow edit Flow edit

Edita imágenes con este modelo de Flow. Basándose en SD3 o FLUX puedes editar cualquier imagen y generar nuevas

FLUX.1-RealismLora

FLUX.1-RealismLora FLUX.1-RealismLora
Ver todos los contenedores -->

¿Quieres aplicar la IA en tu proyecto? Contactame!

¿Quieres entrenar tu modelo con estos datasets?

short-jokes-dataset

Dataset de chistes en inglés

opus100

Dataset con traducciones de inglés a español

netflix_titles

Dataset con películas y series de Netflix

Ver más datasets -->