MCP Chain

Un marco de middleware componible para construir cadenas de servidores MCP sofisticadas, inspirado en Ruby Rack.

GitHub
7
Prueba este MCPPatrocinado

Documentación

MCP Chain

CI codecov PyPI version GitHub release Python versions License

Un framework de middleware componible para construir cadenas de servidores MCP, inspirado en Ruby Rack. MCP Chain te permite crear proxies transparentes que se sitúan entre clientes y servidores MCP, transformando solicitudes y respuestas mediante funciones de Python.

MCP Chain resuelve el problema de añadir preocupaciones transversales (autenticación, registro de actividad, transformación de solicitudes) a servidores MCP existentes sin modificarlos. Utiliza un patrón de proxy transparente donde cada capa de middleware aparece como un servidor MCP estándar para los clientes, mientras reenvía solicitudes a servidores posteriores. El middleware también puede orquestar múltiples llamadas MCP en segundo plano usando IA, transformando APIs granulares en MCPs inteligentes que realizan tareas complejas de múltiples pasos.

Inicio rápido

Instala y ejecuta con uvx - no se requiere configuración:

# 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

Añade a tu mcp.json:

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

Ejemplos

Middleware de Autenticación

Añade autenticación a cualquier 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")

Transformación de Solicitudes/Respuestas

Transforma metadatos y solicitudes:

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))

Cadena de Múltiples Middlewares

Apila autenticación, registro de actividad y transformación:

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

Usa la función serve() directamente:

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)

Arquitectura

MCP Chain utiliza un patrón de middleware funcional donde cada capa transforma solicitudes/respuestas y reenvía a la siguiente capa:

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 capa de middleware:

  1. Recibe solicitudes de la capa anterior (o del cliente)
  2. Transforma la solicitud/metadatos usando diccionarios de Python
  3. Reenvía a la siguiente capa (o al servidor posterior)
  4. Recibe la respuesta de vuelta
  5. Transforma la respuesta según sea necesario
  6. Devuelve a la capa anterior (o al cliente)

Principios Fundamentales:

  • Proxy Transparente: Cada middleware aparece como un servidor MCP estándar para los clientes
  • Procesamiento Basado en Diccionarios: El procesamiento interno usa diccionarios de Python, no cadenas JSON
  • Componible: El middleware puede encadenarse ya que cada capa es un servidor MCP
  • Cero Sobrecarga: Sin serialización/deserialización en la cadena de middleware

Construido sobre el SDK oficial de FastMCP para total cumplimiento del protocolo MCP.

API

Funciones Principales

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)

Construcción de Cadenas

# 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-Detección

El CLI detecta automáticamente variables de cadena en tus archivos de 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(...)

Ejecuta con: uvx mcp-chain filename.py

Desarrollo

Este proyecto fue desarrollado principalmente con asistentes de IA y está diseñado para flujos de trabajo de desarrollo asistidos por IA. La estructura del código está pensada para ser fácilmente comprendida y modificada por herramientas de IA. La carpeta ai/ contiene documentos de contexto y notas de diseño específicamente para asistentes de IA que trabajen en este repositorio.

Instalación

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

Pruebas

# 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

Publicación

uv build && uv publish

Las versiones se publican automáticamente en PyPI mediante GitHub Actions en cada nueva versión.

Opciones de Instalación

# 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