wehub-resource-sync
21 KiB
21 KiB
Análisis Profundo de las Funcionalidades del Protocolo MCP
Esta guía explora funcionalidades avanzadas del protocolo MCP que van más allá del manejo básico de herramientas y recursos. Comprender estas características te ayuda a construir servidores MCP más robustos, amigables para el usuario y listos para producción.
Funcionalidades Cubiertas
- Notificaciones de Progreso - Reportar el progreso para operaciones de larga duración
- Cancelación de Solicitudes - Permitir a los clientes cancelar solicitudes en curso
- Plantillas de Recursos - URIs dinámicas para recursos con parámetros
- Eventos del Ciclo de Vida del Servidor - Inicialización y apagado apropiados
- Control de Registro de Eventos - Configuración del registro en el lado del servidor
- Patrones de Manejo de Errores - Respuestas consistentes ante errores
1. Notificaciones de Progreso
Para operaciones que toman tiempo (procesamiento de datos, descargas de archivos, llamadas API), las notificaciones de progreso mantienen informados a los usuarios.
Cómo Funciona
sequenceDiagram
participant Client
participant Server
Client->>Server: tools/call (operación larga)
Server-->>Client: notificación: progreso 10%
Server-->>Client: notificación: progreso 50%
Server-->>Client: notificación: progreso 90%
Server->>Client: resultado (completo)
Implementación en Python
from mcp.server import Server, NotificationOptions
from mcp.types import ProgressNotification
import asyncio
app = Server("progress-server")
@app.tool()
async def process_large_file(file_path: str, ctx) -> str:
"""Process a large file with progress updates."""
# Obtener tamaño del archivo para cálculo del progreso
file_size = os.path.getsize(file_path)
processed = 0
with open(file_path, 'rb') as f:
while chunk := f.read(8192):
# Procesar fragmento
await process_chunk(chunk)
processed += len(chunk)
# Enviar notificación de progreso
progress = (processed / file_size) * 100
await ctx.send_notification(
ProgressNotification(
progressToken=ctx.request_id,
progress=progress,
total=100,
message=f"Processing: {progress:.1f}%"
)
)
return f"Processed {file_size} bytes"
@app.tool()
async def batch_operation(items: list[str], ctx) -> str:
"""Process multiple items with progress."""
results = []
total = len(items)
for i, item in enumerate(items):
result = await process_item(item)
results.append(result)
# Reportar progreso después de cada elemento
await ctx.send_notification(
ProgressNotification(
progressToken=ctx.request_id,
progress=i + 1,
total=total,
message=f"Processed {i + 1}/{total}: {item}"
)
)
return f"Completed {total} items"
Implementación en TypeScript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
server.setRequestHandler(CallToolSchema, async (request, extra) => {
const { name, arguments: args } = request.params;
if (name === "process_data") {
const items = args.items as string[];
const results = [];
for (let i = 0; i < items.length; i++) {
const result = await processItem(items[i]);
results.push(result);
// Enviar notificación de progreso
await extra.sendNotification({
method: "notifications/progress",
params: {
progressToken: request.id,
progress: i + 1,
total: items.length,
message: `Processing item ${i + 1}/${items.length}`
}
});
}
return { content: [{ type: "text", text: JSON.stringify(results) }] };
}
});
Manejo en el Cliente (Python)
async def handle_progress(notification):
"""Handle progress notifications from server."""
params = notification.params
print(f"Progress: {params.progress}/{params.total} - {params.message}")
# Registrar manejador
session.on_notification("notifications/progress", handle_progress)
# Llamar herramienta (las actualizaciones de progreso llegarán a través del manejador)
result = await session.call_tool("process_large_file", {"file_path": "/data/large.csv"})
2. Cancelación de Solicitudes
Permitir a los clientes cancelar solicitudes que ya no se necesitan o que están tardando demasiado.
Implementación en Python
from mcp.server import Server
from mcp.types import CancelledError
import asyncio
app = Server("cancellable-server")
@app.tool()
async def long_running_search(query: str, ctx) -> str:
"""Search that can be cancelled."""
results = []
try:
for page in range(100): # Buscar a través de muchas páginas
# Verificar si se solicitó la cancelación
if ctx.is_cancelled:
raise CancelledError("Search cancelled by user")
# Simular búsqueda en la página
page_results = await search_page(query, page)
results.extend(page_results)
# Pequeña demora permite verificar cancelaciones
await asyncio.sleep(0.1)
except CancelledError:
# Devolver resultados parciales
return f"Cancelled. Found {len(results)} results before cancellation."
return f"Found {len(results)} total results"
@app.tool()
async def download_file(url: str, ctx) -> str:
"""Download with cancellation support."""
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
total_size = int(response.headers.get('content-length', 0))
downloaded = 0
chunks = []
async for chunk in response.content.iter_chunked(8192):
if ctx.is_cancelled:
return f"Download cancelled at {downloaded}/{total_size} bytes"
chunks.append(chunk)
downloaded += len(chunk)
return f"Downloaded {downloaded} bytes"
Implementación del Contexto de Cancelación
class CancellableContext:
"""Context object that tracks cancellation state."""
def __init__(self, request_id: str):
self.request_id = request_id
self._cancelled = asyncio.Event()
self._cancel_reason = None
@property
def is_cancelled(self) -> bool:
return self._cancelled.is_set()
def cancel(self, reason: str = "Cancelled"):
self._cancel_reason = reason
self._cancelled.set()
async def check_cancelled(self):
"""Raise if cancelled, otherwise continue."""
if self.is_cancelled:
raise CancelledError(self._cancel_reason)
async def sleep_or_cancel(self, seconds: float):
"""Sleep that can be interrupted by cancellation."""
try:
await asyncio.wait_for(
self._cancelled.wait(),
timeout=seconds
)
raise CancelledError(self._cancel_reason)
except asyncio.TimeoutError:
pass # Tiempo de espera normal, continuar
Cancelación del Lado del Cliente
import asyncio
async def search_with_timeout(session, query, timeout=30):
"""Search with automatic cancellation on timeout."""
task = asyncio.create_task(
session.call_tool("long_running_search", {"query": query})
)
try:
result = await asyncio.wait_for(task, timeout=timeout)
return result
except asyncio.TimeoutError:
# Cancelación de solicitud
await session.send_notification({
"method": "notifications/cancelled",
"params": {"requestId": task.request_id, "reason": "Timeout"}
})
return "Search timed out"
3. Plantillas de Recursos
Las plantillas de recursos permiten la construcción dinámica de URIs con parámetros, útil para APIs y bases de datos.
Definiendo Plantillas
from mcp.server import Server
from mcp.types import ResourceTemplate
app = Server("template-server")
@app.list_resource_templates()
async def list_templates() -> list[ResourceTemplate]:
"""Return available resource templates."""
return [
ResourceTemplate(
uriTemplate="db://users/{user_id}",
name="User Profile",
description="Fetch user profile by ID",
mimeType="application/json"
),
ResourceTemplate(
uriTemplate="api://weather/{city}/{date}",
name="Weather Data",
description="Historical weather for city and date",
mimeType="application/json"
),
ResourceTemplate(
uriTemplate="file://{path}",
name="File Content",
description="Read file at given path",
mimeType="text/plain"
)
]
@app.read_resource()
async def read_resource(uri: str) -> str:
"""Read resource, expanding template parameters."""
# Analizar el URI para extraer parámetros
if uri.startswith("db://users/"):
user_id = uri.split("/")[-1]
return await fetch_user(user_id)
elif uri.startswith("api://weather/"):
parts = uri.replace("api://weather/", "").split("/")
city, date = parts[0], parts[1]
return await fetch_weather(city, date)
elif uri.startswith("file://"):
path = uri.replace("file://", "")
return await read_file(path)
raise ValueError(f"Unknown resource URI: {uri}")
Implementación en TypeScript
server.setRequestHandler(ListResourceTemplatesSchema, async () => {
return {
resourceTemplates: [
{
uriTemplate: "github://repos/{owner}/{repo}/issues/{issue_number}",
name: "GitHub Issue",
description: "Fetch a specific GitHub issue",
mimeType: "application/json"
},
{
uriTemplate: "db://tables/{table}/rows/{id}",
name: "Database Row",
description: "Fetch a row from a database table",
mimeType: "application/json"
}
]
};
});
server.setRequestHandler(ReadResourceSchema, async (request) => {
const uri = request.params.uri;
// Analizar URI de problema de GitHub
const githubMatch = uri.match(/^github:\/\/repos\/([^/]+)\/([^/]+)\/issues\/(\d+)$/);
if (githubMatch) {
const [_, owner, repo, issueNumber] = githubMatch;
const issue = await fetchGitHubIssue(owner, repo, parseInt(issueNumber));
return {
contents: [{
uri,
mimeType: "application/json",
text: JSON.stringify(issue, null, 2)
}]
};
}
throw new Error(`Unknown resource URI: ${uri}`);
});
4. Eventos del Ciclo de Vida del Servidor
El manejo adecuado de la inicialización y el apagado asegura una gestión limpia de los recursos.
Gestión del Ciclo de Vida en Python
from mcp.server import Server
from contextlib import asynccontextmanager
app = Server("lifecycle-server")
# Estado compartido
db_connection = None
cache = None
@asynccontextmanager
async def lifespan(server: Server):
"""Manage server lifecycle."""
global db_connection, cache
# Inicio
print("🚀 Server starting...")
db_connection = await create_database_connection()
cache = await create_cache_client()
print("✅ Resources initialized")
yield # El servidor se ejecuta aquí
# Apagado
print("🛑 Server shutting down...")
await db_connection.close()
await cache.close()
print("✅ Resources cleaned up")
app = Server("lifecycle-server", lifespan=lifespan)
@app.tool()
async def query_database(sql: str) -> str:
"""Use the shared database connection."""
result = await db_connection.execute(sql)
return str(result)
Ciclo de Vida en TypeScript
import { Server } from "@modelcontextprotocol/sdk/server/index.js";
class ManagedServer {
private server: Server;
private dbConnection: DatabaseConnection | null = null;
constructor() {
this.server = new Server({
name: "lifecycle-server",
version: "1.0.0"
});
this.setupHandlers();
}
async start() {
// Inicializar recursos
console.log("🚀 Server starting...");
this.dbConnection = await createDatabaseConnection();
console.log("✅ Database connected");
// Iniciar servidor
await this.server.connect(transport);
}
async stop() {
// Limpiar recursos
console.log("🛑 Server shutting down...");
if (this.dbConnection) {
await this.dbConnection.close();
}
await this.server.close();
console.log("✅ Cleanup complete");
}
private setupHandlers() {
this.server.setRequestHandler(CallToolSchema, async (request) => {
// Usar this.dbConnection de forma segura
// ...
});
}
}
// Uso con apagado controlado
const server = new ManagedServer();
process.on('SIGINT', async () => {
await server.stop();
process.exit(0);
});
await server.start();
5. Control de Registro de Eventos
MCP soporta niveles de registro en el lado del servidor que los clientes pueden controlar.
Implementando Niveles de Registro
from mcp.server import Server
from mcp.types import LoggingLevel
import logging
app = Server("logging-server")
# Mapear niveles MCP a niveles de registro de Python
LEVEL_MAP = {
LoggingLevel.DEBUG: logging.DEBUG,
LoggingLevel.INFO: logging.INFO,
LoggingLevel.WARNING: logging.WARNING,
LoggingLevel.ERROR: logging.ERROR,
}
logger = logging.getLogger("mcp-server")
@app.set_logging_level()
async def set_logging_level(level: LoggingLevel) -> None:
"""Handle client request to change logging level."""
python_level = LEVEL_MAP.get(level, logging.INFO)
logger.setLevel(python_level)
logger.info(f"Logging level set to {level}")
@app.tool()
async def debug_operation(data: str) -> str:
"""Tool with various logging levels."""
logger.debug(f"Processing data: {data}")
try:
result = process(data)
logger.info(f"Successfully processed: {result}")
return result
except Exception as e:
logger.error(f"Processing failed: {e}")
raise
Envío de Mensajes de Registro al Cliente
@app.tool()
async def complex_operation(input: str, ctx) -> str:
"""Operation that logs to client."""
# Enviar notificación de registro al cliente
await ctx.send_log(
level="info",
message=f"Starting complex operation with input: {input}"
)
# Realizar trabajo...
result = await do_work(input)
await ctx.send_log(
level="debug",
message=f"Operation complete, result size: {len(result)}"
)
return result
6. Patrones de Manejo de Errores
El manejo consistente de errores mejora la depuración y la experiencia del usuario.
Códigos de Error MCP
from mcp.types import McpError, ErrorCode
class ToolError(McpError):
"""Base class for tool errors."""
pass
class ValidationError(ToolError):
"""Invalid input parameters."""
def __init__(self, message: str):
super().__init__(ErrorCode.INVALID_PARAMS, message)
class NotFoundError(ToolError):
"""Requested resource not found."""
def __init__(self, resource: str):
super().__init__(ErrorCode.INVALID_REQUEST, f"Not found: {resource}")
class PermissionError(ToolError):
"""Access denied."""
def __init__(self, action: str):
super().__init__(ErrorCode.INVALID_REQUEST, f"Permission denied: {action}")
class InternalError(ToolError):
"""Internal server error."""
def __init__(self, message: str):
super().__init__(ErrorCode.INTERNAL_ERROR, message)
Respuestas de Error Estructuradas
@app.tool()
async def safe_operation(input: str) -> str:
"""Tool with comprehensive error handling."""
# Validar entrada
if not input:
raise ValidationError("Input cannot be empty")
if len(input) > 10000:
raise ValidationError(f"Input too large: {len(input)} chars (max 10000)")
try:
# Verificar permisos
if not await check_permission(input):
raise PermissionError(f"read {input}")
# Realizar operación
result = await perform_operation(input)
if result is None:
raise NotFoundError(input)
return result
except ConnectionError as e:
raise InternalError(f"Database connection failed: {e}")
except TimeoutError as e:
raise InternalError(f"Operation timed out: {e}")
except Exception as e:
# Registrar errores inesperados
logger.exception(f"Unexpected error in safe_operation")
raise InternalError(f"Unexpected error: {type(e).__name__}")
Manejo de Errores en TypeScript
import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";
function validateInput(data: unknown): asserts data is ValidInput {
if (typeof data !== "object" || data === null) {
throw new McpError(
ErrorCode.InvalidParams,
"Input must be an object"
);
}
// Más validación...
}
server.setRequestHandler(CallToolSchema, async (request) => {
try {
validateInput(request.params.arguments);
const result = await performOperation(request.params.arguments);
return {
content: [{ type: "text", text: JSON.stringify(result) }]
};
} catch (error) {
if (error instanceof McpError) {
throw error; // Ya es un error MCP
}
// Convertir otros errores
if (error instanceof NotFoundError) {
throw new McpError(ErrorCode.InvalidRequest, error.message);
}
// Error desconocido
console.error("Unexpected error:", error);
throw new McpError(
ErrorCode.InternalError,
"An unexpected error occurred"
);
}
});
Funcionalidades Experimentales (MCP 2025-11-25)
Estas funcionalidades están marcadas como experimentales en la especificación:
Tareas (Operaciones de Larga Duración)
# Las tareas permiten rastrear operaciones de larga duración con estado
@app.task()
async def training_task(model_id: str, data_path: str, ctx) -> str:
"""Long-running ML training task."""
# Reportar tarea iniciada
await ctx.report_status("running", "Initializing training...")
# Bucle de entrenamiento
for epoch in range(100):
await train_epoch(model_id, data_path, epoch)
await ctx.report_status(
"running",
f"Training epoch {epoch + 1}/100",
progress=epoch + 1,
total=100
)
await ctx.report_status("completed", "Training finished")
return f"Model {model_id} trained successfully"
Anotaciones de Herramientas
# Las anotaciones proporcionan metadatos sobre el comportamiento de la herramienta
@app.tool(
annotations={
"destructive": False, # No modifica los datos
"idempotent": True, # Seguro para reintentar
"timeout_seconds": 30, # Duración máxima esperada
"requires_approval": False # No se necesita aprobación del usuario
}
)
async def safe_query(query: str) -> str:
"""A read-only database query tool."""
return await execute_read_query(query)
Qué Sigue
- Módulo 8 - Mejores Prácticas
- 5.14 - Ingeniería de Contexto
- Registro de Cambios de la Especificación MCP
Recursos Adicionales
- Especificación MCP 2025-11-25
- Códigos de Error JSON-RPC 2.0
- Ejemplos SDK Python
- Ejemplos SDK TypeScript
Aviso Legal: Este documento ha sido traducido utilizando el servicio de traducción automática Co-op Translator. Aunque nos esforzamos por la precisión, tenga en cuenta que las traducciones automáticas pueden contener errores o inexactitudes. El documento original en su idioma nativo debe considerarse la fuente autorizada. Para información crítica, se recomienda una traducción profesional realizada por humanos. No nos hacemos responsables de cualquier malentendido o interpretación errónea derivada del uso de esta traducción.