MCP Chain

Um framework de middleware componível para construir cadeias sofisticadas de servidores MCP, inspirado no Ruby Rack.

GitHub
7
Experimente este MCPPatrocinado

Documentação

Tradução

MCP Chain

CI codecov PyPI version GitHub release Python versions License

Um framework de middleware componível para construir cadeias de servidores MCP, inspirado no Ruby Rack. O MCP Chain permite criar proxies transparentes que ficam entre clientes e servidores MCP, transformando requisições e respostas usando funções Python.

O MCP Chain resolve o problema de adicionar preocupações transversais (autenticação, registro de logs, transformação de requisições) a servidores MCP existentes sem modificá-los. Ele usa um padrão de proxy transparente onde cada camada de middleware aparece como um servidor MCP padrão para os clientes, enquanto encaminha requisições para servidores downstream. O middleware também pode orquestrar múltiplas chamadas MCP nos bastidores usando IA, transformando APIs granulares em MCPs inteligentes que executam tarefas complexas de múltiplas etapas.

Início Rápido

Instale e execute com uvx - nenhuma configuração necessária:

# cli_server.py
from mcp_chain import mcp_chain, CLIMCPServer

cli_server = CLIMCPServer(
    name="dev-tools",
    commands=["git", "ls", "grep"],
    descriptions={
        "git": "Git version control operations",
        "ls": "List directory contents", 
        "grep": "Search text patterns"
    }
)

# Auto-detected by CLI
chain = mcp_chain().then(cli_server)

uvx mcp-chain cli_server.py

Adicione ao seu mcp.json:

{
  "mcpServers": {
    "dev-tools": {
      "command": "uvx",
      "args": ["mcp-chain", "cli_server.py"]
    }
  }
}

Exemplos

Middleware de Autenticação

Adicione autenticação a qualquer servidor MCP:

from mcp_chain import mcp_chain, ExternalMCPServer, serve

def require_auth(next_server, request_dict):
    if not request_dict.get("auth_token"):
        return {"error": "Authentication required", "code": 401}
    return next_server.handle_request(request_dict)

chain = (mcp_chain()
         .then(None, require_auth)
         .then(ExternalMCPServer("postgres", "postgres-mcp")))

serve(chain, name="Authenticated Postgres")

Transformação de Requisição/Resposta

Transforme metadados e requisições:

from mcp_chain import mcp_chain, CLIMCPServer

def add_company_context(next_server, metadata_dict):
    metadata = next_server.get_metadata()
    for tool in metadata.get("tools", []):
        tool["description"] = f"ACME Corp: {tool.get('description', '')}"
    return metadata

def add_headers(next_server, request_dict):
    request_dict["headers"] = {"X-Company": "ACME"}
    response = next_server.handle_request(request_dict)
    response["processed_by"] = "acme-proxy"
    return response

cli_server = CLIMCPServer(name="tools", commands=["git", "docker"])

chain = (mcp_chain()
         .then(add_company_context, add_headers)
         .then(cli_server))

Cadeia de Múltiplos Middlewares

Empilhe autenticação, registro de logs e transformação:

import logging
from mcp_chain import mcp_chain, ExternalMCPServer, serve

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("mcp-chain")

def auth_middleware(next_server, request_dict):
    if not request_dict.get("auth_token"):
        return {"error": "Authentication required", "code": 401}
    return next_server.handle_request(request_dict)

def logging_middleware(next_server, request_dict):
    logger.info(f"Request: {request_dict.get('method')}")
    response = next_server.handle_request(request_dict)
    logger.info(f"Response: {response.get('result', 'error')}")
    return response

def context_middleware(next_server, metadata_dict):
    metadata = next_server.get_metadata()
    for tool in metadata.get("tools", []):
        tool["description"] = f"Enterprise: {tool.get('description', '')}"
    return metadata

chain = (mcp_chain()
         .then(context_middleware, auth_middleware)
         .then(None, logging_middleware)
         .then(ExternalMCPServer("postgres", "postgres-mcp")))

serve(chain, name="Enterprise Postgres")

Uso Programático

Use a função serve() diretamente:

from mcp_chain import mcp_chain, CLIMCPServer, serve

def rate_limit_middleware(next_server, request_dict):
    # Add rate limiting logic
    return next_server.handle_request(request_dict)

cli_server = CLIMCPServer(name="secure-tools", commands=["git"])
chain = mcp_chain().then(None, rate_limit_middleware).then(cli_server)

serve(chain, name="Rate Limited Tools", port=8000)

Arquitetura

O MCP Chain usa um padrão de middleware funcional onde cada camada transforma requisições/respostas e encaminha para a próxima camada:

graph TD
  A["MCP Client"] --> B["FastMCP"]
  B --> C["mcp_chain()"]
  C --> D1["middleware_1"]
  D1 --> D2["middleware_2"]
  D2 --> E["downstream_server"]
  E -- "response" --> D2
  D2 -- "response" --> D1
  D1 -- "response" --> C
  C -- "response" --> B
  B -- "response" --> A

Cada camada de middleware:

  1. Recebe requisições da camada anterior (ou do cliente)
  2. Transforma a requisição/metadados usando dicionários Python
  3. Encaminha para a próxima camada (ou servidor downstream)
  4. Recebe a resposta de volta
  5. Transforma a resposta conforme necessário
  6. Retorna para a camada anterior (ou cliente)

Princípios Fundamentais:

  • Proxy Transparente: Cada middleware aparece como um servidor MCP padrão para os clientes
  • Processamento Baseado em Dicionários: O processamento interno usa dicionários Python, não strings JSON
  • Componível: Middlewares podem ser encadeados, pois cada camada é um servidor MCP
  • Zero Overhead: Sem serialização/desserialização na cadeia de middleware

Construído sobre o SDK oficial FastMCP para conformidade completa com o protocolo MCP.

API

Funções Principais

from mcp_chain import mcp_chain, serve, CLIMCPServer, ExternalMCPServer

# Create a chain
chain = mcp_chain()

# Add middleware layers
chain = chain.then(metadata_transformer, request_transformer)
chain = chain.then(downstream_server)

# Start server
serve(chain, name="My Server", port=8000)

Construção de Cadeias

# Metadata transformer (transforms server capabilities)
def metadata_transformer(next_server, metadata_dict):
    metadata = next_server.get_metadata()
    # Transform metadata dict and return
    return metadata

# Request transformer (transforms requests/responses)  
def request_transformer(next_server, request_dict):
    # Transform request dict
    response = next_server.handle_request(request_dict)
    # Transform response dict and return
    return response

# Add to chain
chain = mcp_chain().then(metadata_transformer, request_transformer)

Servidores Integrados

# CLI server - exposes command-line tools as MCP tools
cli_server = CLIMCPServer(
    name="my-tools",
    commands=["git", "docker", "npm"],
    descriptions={
        "git": "Git operations",
        "docker": "Container management",
        "npm": "Package management"
    }
)

# External server proxy
external_server = ExternalMCPServer("server-name", "command-to-run")

Auto-Detecção

A CLI detecta automaticamente variáveis de cadeia em seus arquivos Python:

# Any of these variable names work:
chain = mcp_chain().then(...)
my_chain = mcp_chain().then(...)  
server_chain = mcp_chain().then(...)
proxy = mcp_chain().then(...)

Execute com: uvx mcp-chain filename.py

Desenvolvimento

Este projeto foi desenvolvido principalmente com assistentes de IA e é projetado para fluxos de trabalho de desenvolvimento assistidos por IA. A base de código é estruturada para ser facilmente compreendida e modificada por ferramentas de IA. A pasta ai/ contém documentos de contexto e notas de design especificamente para assistentes de IA que trabalham neste repositório.

Instalação

# Development install
git clone https://github.com/ronie-uliana/mcp-chain
cd mcp-chain
uv sync

Testes

# Fast unit tests
uv run pytest tests/ -m "not integration" -v

# Integration tests (with timeout protection)
timeout 30 uv run pytest tests/ -m integration -v

# All tests
timeout 45 uv run pytest tests/ -v

# Local CI pipeline
./scripts/test-ci.sh

Publicação

uv build && uv publish

As versões são publicadas automaticamente no PyPI via GitHub Actions em novos lançamentos.

Opções de Instalação

# Recommended: Run with uvx (no installation)
uvx mcp-chain my_chain.py

# Install from PyPI
pip install mcp-chain

# Run installed version
python -m mcp_chain my_chain.py
# or
mcp-chain my_chain.py