mcp-backpressure

mcp-backpressure

Middleware de backpressure e controle de concorrência para servidores MCP FastMCP.

Problema: LLMs podem gerar centenas de chamadas de ferramentas paralelas, causando esgotamento de recursos, falhas no servidor e nenhum feedback estruturado para os clientes tentarem novamente.

Solução: Middleware que limita execuções concorrentes, enfileira solicitações excedentes com timeout e retorna erros estruturados de sobrecarga em JSON-RPC.

Início 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)
))

Instalação

pip install mcp-backpressure

Recursos

• Limitação de concorrência: Controle baseado em semáforo de execuções paralelas
• Fila limitada: Fila FIFO opcional com tamanho configurável
• Timeout de fila: Timeout automático para solicitações enfileiradas com limpeza
• Erros estruturados: Erros de sobrecarga compatíveis com JSON-RPC com métricas detalhadas
• Métricas: Contadores em tempo real para solicitações ativas, enfileiradas e rejeitadas
• Hook de callback: Notificação opcional em cada evento de sobrecarga
• Zero dependências: Requer apenas FastMCP e Python 3.10+

Uso

Configuração 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	Padrão	Descrição
max_concurrent	int	obrigatório	Número máximo de execuções de ferramentas concorrentes. Deve ser >= 1.
queue_size	int	0	Tamanho máximo da fila para solicitações em espera. Defina como 0 para rejeitar imediatamente quando o limite for atingido.
queue_timeout	float	30.0	Tempo máximo (em segundos) que uma solicitação pode aguardar na fila antes do timeout. Deve ser > 0.
overload_error_code	int	-32001	Código de erro JSON-RPC retornado quando o servidor está sobrecarregado.
on_overload	Callable	None	Callback opcional (error: OverloadError) -> None invocado em cada sobrecarga.

Tratamento de Erros

Quando o servidor está sobrecarregado, as solicitações são rejeitadas com um erro estruturado 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
  }
}

Motivos de Sobrecarga

Motivo	Descrição
concurrency_limit	Todos os slots de execução ocupados e nenhuma fila configurada (queue_size=0)
queue_full	Todos os slots de execução e slots de fila estão ocupados
queue_timeout	A solicitação aguardou na fila por mais tempo que queue_timeout

Métricas

Obtenha métricas em tempo real do 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 assíncronos, use await middleware.get_metrics_async().

Hook de Callback

Registre um 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,
)

Exemplos

Servidor Simples

Consulte examples/simple_server.py para um servidor FastMCP mínimo com backpressure.

Simulação de Carga

Execute examples/load_simulation.py para ver o comportamento do backpressure sob carga concorrente pesada:

python examples/load_simulation.py

Isso simula 30 solicitações concorrentes contra um servidor limitado a 5 execuções concorrentes com uma fila de 10, demonstrando como o middleware lida com sobrecarga.

Como Funciona

O middleware fornece limitação em dois níveis:

1. Semáforo (max_concurrent): Controla execuções ativas
2. Fila limitada (queue_size): Mantém solicitações em espera com timeout

Fluxo de solicitação:

• Se um slot de execução estiver disponível → executa imediatamente
• Se os slots de execução estiverem cheios e a fila não estiver cheia → aguarda na fila com timeout
• Se a fila estiver cheia → rejeita com queue_full
• Se houver timeout na fila → rejeita com queue_timeout

Invariantes (garantidos sob todas as condições):

• active <= max_concurrent SEMPRE
• queued <= queue_size SEMPRE
• Cancelamento libera slots corretamente e decrementa contadores
• Timeout de fila remove o item da fila

Desenvolvimento

Executando Testes

python -m pytest tests/ -v

Linting

ruff check src/ tests/

Racional de Design

Esta biblioteca surgiu de python-sdk #1698 (fechado como "não planejado"). Principais decisões de design:

• Apenas limites globais (v0.1): Limites por cliente e por ferramenta adiados para v0.2+
• Contadores simples: Sem dependências de Prometheus/OTEL por padrão
• Erros JSON-RPC: Segue as convenções do protocolo MCP
• Tempo monotônico: Timeouts de fila usam time.monotonic() para confiabilidade

Licença

MIT

Contribuição

Contribuições são bem-vindas! Por favor, abra uma issue antes de enviar PRs.

Changelog

Consulte CHANGELOG.md