mcp-backpressure

mcp-backpressure

Middleware de control de contrapresión y concurrencia para servidores MCP de FastMCP.

Problema: Los LLMs pueden generar cientos de llamadas a herramientas en paralelo, lo que provoca agotamiento de recursos, caídas del servidor y falta de retroalimentación estructurada para que los clientes puedan reintentar.

Solución: Middleware que limita las ejecuciones concurrentes, pone en cola las solicitudes excesivas con tiempo de espera y devuelve errores estructurados de sobrecarga compatibles con JSON-RPC.

Inicio rápido

from fastmcp import FastMCP
from mcp_backpressure import BackpressureMiddleware

mcp = FastMCP("MyServer")
mcp.add_middleware(BackpressureMiddleware(
    max_concurrent=5,      # Max parallel executions
    queue_size=10,         # Bounded queue for waiting requests
    queue_timeout=30.0,    # Queue wait timeout (seconds)
))

Instalación

pip install mcp-backpressure

Características

• Límite de concurrencia: Control basado en semáforos de ejecuciones paralelas
• Cola acotada: Cola FIFO opcional con tamaño configurable
• Tiempo de espera en cola: Tiempo de espera automático para solicitudes en cola con limpieza
• Errores estructurados: Errores de sobrecarga compatibles con JSON-RPC con métricas detalladas
• Métricas: Contadores en tiempo real de solicitudes activas, en cola y rechazadas
• Hook de callback: Notificación opcional en cada evento de sobrecarga
• Cero dependencias: Solo requiere FastMCP y Python 3.10+

Uso

Configuración básica

from mcp_backpressure import BackpressureMiddleware

mcp.add_middleware(BackpressureMiddleware(
    max_concurrent=5,           # Required: max parallel tool executions
    queue_size=10,              # Optional: bounded queue (0 = no queue)
    queue_timeout=30.0,         # Optional: seconds to wait in queue
    overload_error_code=-32001, # Optional: JSON-RPC error code
    on_overload=callback,       # Optional: called on each overload
))

Parámetros

Parámetro	Tipo	Predeterminado	Descripción
max_concurrent	int	obligatorio	Número máximo de ejecuciones de herramientas concurrentes. Debe ser >= 1.
queue_size	int	0	Tamaño máximo de la cola para solicitudes en espera. Establecer en 0 para rechazar inmediatamente cuando se alcance el límite.
queue_timeout	float	30.0	Tiempo máximo (en segundos) que una solicitud puede esperar en la cola antes de agotar el tiempo. Debe ser > 0.
overload_error_code	int	-32001	Código de error JSON-RPC devuelto cuando el servidor está sobrecargado.
on_overload	Callable	None	Callback opcional (error: OverloadError) -> None invocado en cada sobrecarga.

Manejo de errores

Cuando el servidor está sobrecargado, las solicitudes se rechazan con un error estructurado JSON-RPC:

{
  "code": -32001,
  "message": "SERVER_OVERLOADED",
  "data": {
    "reason": "queue_full",
    "active": 5,
    "queued": 10,
    "max_concurrent": 5,
    "queue_size": 10,
    "queue_timeout_ms": 30000,
    "retry_after_ms": 1000
  }
}

Razones de sobrecarga

Razón	Descripción
concurrency_limit	Todos los espacios de ejecución están ocupados y no hay cola configurada (queue_size=0)
queue_full	Todos los espacios de ejecución y de cola están ocupados
queue_timeout	La solicitud esperó en la cola más tiempo que queue_timeout

Métricas

Obtén métricas en tiempo real del middleware:

metrics = middleware.get_metrics()  # Synchronous

print(f"Active: {metrics.active}")
print(f"Queued: {metrics.queued}")
print(f"Total rejected: {metrics.total_rejected}")
print(f"Rejected (concurrency): {metrics.rejected_concurrency_limit}")
print(f"Rejected (queue full): {metrics.rejected_queue_full}")
print(f"Rejected (timeout): {metrics.rejected_queue_timeout}")

Para contextos asíncronos, usa await middleware.get_metrics_async().

Hook de callback

Registra un callback para ser notificado de cada evento de sobrecarga:

def on_overload(error: OverloadError):
    print(f"OVERLOAD: {error.reason} (active={error.active})")
    # Log to monitoring system, update metrics, etc.

middleware = BackpressureMiddleware(
    max_concurrent=5,
    queue_size=10,
    on_overload=on_overload,
)

Ejemplos

Servidor simple

Consulta examples/simple_server.py para un servidor FastMCP mínimo con contrapresión.

Simulación de carga

Ejecuta examples/load_simulation.py para ver el comportamiento de la contrapresión bajo carga concurrente intensa:

python examples/load_simulation.py

Esto simula 30 solicitudes concurrentes contra un servidor limitado a 5 ejecuciones concurrentes con una cola de 10, demostrando cómo el middleware maneja la sobrecarga.

Cómo funciona

El middleware proporciona un límite de dos niveles:

1. Semáforo (max_concurrent): Controla las ejecuciones activas
2. Cola acotada (queue_size): Mantiene las solicitudes en espera con tiempo de espera

Flujo de solicitudes:

• Si hay un espacio de ejecución disponible → ejecutar inmediatamente
• Si los espacios de ejecución están llenos y la cola no está llena → esperar en la cola con tiempo de espera
• Si la cola está llena → rechazar con queue_full
• Si se agota el tiempo en la cola → rechazar con queue_timeout

Invariantes (garantizados bajo todas las condiciones):

• active <= max_concurrent SIEMPRE
• queued <= queue_size SIEMPRE
• La cancelación libera correctamente los espacios y decrementa los contadores
• El tiempo de espera en la cola elimina el elemento de la cola

Desarrollo

Ejecutar pruebas

python -m pytest tests/ -v

Linting

ruff check src/ tests/

Justificación del diseño

Esta biblioteca surgió de python-sdk #1698 (cerrado como "no planificado"). Decisiones clave de diseño:

• Solo límites globales (v0.1): Los límites por cliente y por herramienta se posponen a v0.2+
• Contadores simples: Sin dependencias de Prometheus/OTEL por defecto
• Errores JSON-RPC: Sigue las convenciones del protocolo MCP
• Tiempo monotónico: Los tiempos de espera en cola usan time.monotonic() para mayor fiabilidad

Licencia

MIT

Contribuciones

¡Las contribuciones son bienvenidas! Por favor, abre un issue antes de enviar pull requests.

Registro de cambios

Consulta CHANGELOG.md