chuk-mcp

Un cliente de Python para el Protocolo de Contexto de Modelo (MCP), un estándar abierto para conectar asistentes de IA a datos y herramientas externas.

Documentación

chuk-mcp

PyPI version PyPI - Downloads Python Version Code style: ruff License

Una implementación ligera y minimalista del Protocolo de Contexto de Modelos (MCP) en Python.

Lleva soporte de primera clase para el protocolo MCP a Python: ligero, asíncrono y preciso según la especificación desde el primer día.

Requiere Python 3.11+

chuk-mcp te ofrece una implementación limpia, tipada y agnóstica al transporte tanto para clientes como servidores MCP. Se centra en la superficie del protocolo (mensajes, tipos, versionado, transportes) y deja la orquestación, las interfaces de usuario y los marcos de agentes a otras capas.

✳️ Qué es esto: una biblioteca de cumplimiento de protocolo con ayudantes ergonómicos para clientes y servidores.

⛔ Qué no es: un runtime de chatbot, un motor de flujos de trabajo ni un marco de aplicación dogmático.

Arquitectura: Dónde encaja chuk-mcp

Resumen de la pila

┌──────────────────────────────────────┐
│   Your AI Application                │
│   (Claude, GPT, custom agents)       │
└────────────┬─────────────────────────┘
             │ MCP Protocol
             ▼
┌──────────────────────────────────────┐
│   chuk-mcp Client                    │  ← You are here
│   • Protocol compliance              │
│   • Transport (stdio/Streamable HTTP)│
│   • Type-safe messages               │
│   • Capability negotiation           │
└────────────┬─────────────────────────┘
             │ MCP Protocol
             ▼
┌──────────────────────────────────────┐
│   chuk-mcp Server (optional)         │
│   • Protocol handlers                │
│   • Tool/Resource registration       │
│   • Session management               │
└────────────┬─────────────────────────┘
             │
             ▼
┌──────────────────────────────────────┐
│   Your Tools & Resources             │
│   (databases, APIs, files, etc)      │
└──────────────────────────────────────┘

chuk-mcp proporciona la capa de protocolo: conecta aplicaciones de IA a herramientas y fuentes de datos usando el protocolo MCP estándar.

Arquitectura interna

La biblioteca en sí está organizada en capas que puedes usar en diferentes niveles de abstracción:

┌─────────────────────────────────────────┐
│              CLI & Demo Layer           │  __main__.py, demos/
├─────────────────────────────────────────┤
│             Client/Server API           │  High-level abstractions
├─────────────────────────────────────────┤
│            Protocol Layer               │  Messages, types, features
├─────────────────────────────────────────┤
│            Transport Layer              │  stdio, Streamable HTTP
├─────────────────────────────────────────┤
│             Base Layer                  │  Pydantic fallback, config
└─────────────────────────────────────────┘

Detalles de las capas:

CapaPropósitoUso
CLI y DemoUtilidades integradas y demostracionesOpcional: usa la capa de protocolo directamente
API de Cliente/ServidorAbstracciones de alto nivel para interacciones cliente-servidorOpcional: puede usar la capa de protocolo directamente
Capa de ProtocoloDefiniciones de mensajes, manejo tipado de solicitudes/respuestas, negociación de capacidadesNúcleo: implementa la especificación MCP
Capa de TransporteImplementaciones de transporte conectables (stdio, HTTP transmisible)Elige según el despliegue
Capa BaseRespaldo de Pydantic, configuración compartida, adaptadores de tiposFundación: automática

La mayoría de los usuarios trabajan con la Capa de Protocolo (funciones send_*) y la Capa de Transporte (clientes stdio/HTTP), usando opcionalmente la API de Cliente/Servidor para abstracciones de mayor nivel.


Tabla de Contenidos


¿Por qué chuk-mcp?

  • Protocolo primero: Se centra en mensajes MCP, tipos y negociación de capacidades: spec.modelcontextprotocol.io
  • Cliente + Servidor: Soporte completo para construir tanto clientes como servidores MCP
  • Tipado: Anotaciones de tipo completas; modelos Pydantic opcionales cuando están disponibles
  • Agnóstico al transporte: stdio por defecto, HTTP transmisible (NDJSON) para servidores remotos, fácilmente extensible
  • Asíncrono primero: Construido sobre AnyIO; intégrate con anyio.run(...) o tu bucle existente
  • Pequeño y enfocado: Sin orquestación pesada ni suposiciones de agentes
  • Capa de protocolo limpia: Los errores fallan rápido sin reintentos: trae tu propia estrategia de manejo de errores
  • Confiable: Errores claros, enlaces de registro estructurados, componible con capas de reintento/caché
  • ⚡ Alto rendimiento: Sobrecarga del protocolo en el rango de 2-5 ms; JSON rápido opcional para serialización 4 veces más rápida. Consulta Rendimiento del Protocolo para benchmarks detallados

Rendimiento del Protocolo

chuk-mcp está diseñado para mantener la sobrecarga del protocolo MCP en el rango de 2-5 ms, de modo que el costo de usar herramientas esté dominado por las herramientas mismas, no por el protocolo.

Por qué es rápido:

  • Cero dependencias pesadas (solo núcleo AnyIO)
  • stdio y HTTP NDJSON nativos asíncronos
  • Sin ejecución de herramientas dentro de la biblioteca
  • Ruta rápida opcional de orjson ([fast-json])

💡 Para cifras de concurrencia y capacidad, consulta Escalado y Concurrencia.

⚡ Benchmarks de Latencia

Sobrecarga del protocolo (mediciones típicas en hardware moderno):

  • Inicializar → Lista de Herramientas: 2-3 ms
  • Ida y Vuelta de Llamada de Herramienta: < 5 ms de sobrecarga (más allá del tiempo real de ejecución de la herramienta)
  • Transmisión: Sobrecarga casi nula gracias a los límites de fragmentos NDJSON

Benchmarks ejecutados en macOS (Darwin 24.6.0), Python 3.11: consulta benchmarks/PERFORMANCE_REPORT.md para el entorno y comandos exactos.

🚀 Serialización JSON (Ruta Rápida Opcional)

Instala con [fast-json] para operaciones JSON ~4 veces más rápidas usando orjson:

  • Serialización: ~6 veces más rápida
  • Deserialización: ~2 veces más rápida
  • Ida y vuelta: ~4 veces más rápida
pip install "chuk-mcp[fast-json]"  # Automatic with graceful fallback

Números de benchmark de benchmarks/json_performance.py comparando orjson vs json de la biblioteca estándar en mensajes MCP realistas.

🎯 Casos de Uso Ideales

Esto hace que chuk-mcp sea perfecto para:

  • Llamadas de herramientas de alta frecuencia: sobrecarga mínima por solicitud
  • Agentes en tiempo real: latencia de protocolo inferior a 5 ms
  • Interfaces de usuario en streaming: sobrecarga casi nula de fragmentos NDJSON
  • Procesadores de herramientas: lo suficientemente rápidos para ser transparentes
  • Entornos WASM/edge: huella mínima
  • Cargas de trabajo de alto rendimiento: probado a escala (consulta Escalado y Concurrencia)

De un Vistazo

Pruébalo ahora:

# Install an example MCP server
uv tool install mcp-server-sqlite

# Run the quick-start example
uv run python examples/quickstart_sqlite.py

Hola Mundo

Un servidor MCP mínimo funcional en ~10 líneas:

# hello_mcp.py
import anyio
from chuk_mcp.server import MCPServer, run_stdio_server
from chuk_mcp.protocol.types import ServerCapabilities, ToolCapabilities

async def main():
    server = MCPServer("hello", "1.0", ServerCapabilities(tools=ToolCapabilities()))

    async def handle_tools_list(message, session_id):
        return server.protocol_handler.create_response(
            message.id,
            {"tools": [{"name": "hello", "description": "Say hi", "inputSchema": {"type": "object"}}]}
        ), None

    server.protocol_handler.register_method("tools/list", handle_tools_list)
    await run_stdio_server(server)

anyio.run(main)

Ejecútalo: uv run python hello_mcp.py — ¡o conecta cualquier cliente MCP vía stdio!


Stdio (procesos locales):

# Connect to an MCP server via stdio and list tools
import anyio
from chuk_mcp import StdioServerParameters, stdio_client
from chuk_mcp.protocol.messages import send_initialize
from chuk_mcp.protocol.messages.tools import send_tools_list

async def main():
    params = StdioServerParameters(command="uvx", args=["mcp-server-sqlite", "--db-path", "example.db"])
    async with stdio_client(params) as (read, write):
        init = await send_initialize(read, write)
        tools = await send_tools_list(read, write)
        print("Server:", init.serverInfo.name)
        print("Tools:", [t.name for t in tools.tools])

anyio.run(main)

HTTP transmisible (servidores remotos):

# Local dev (plain HTTP)
import anyio
from chuk_mcp.transports.http import http_client, HttpClientParameters
from chuk_mcp.protocol.messages import send_initialize

async def main():
    params = HttpClientParameters(
        url="http://localhost:8989/mcp",
        timeout_s=30,
        headers={"Authorization": "Bearer <token>"}
    )
    async with http_client(params) as (read, write):
        init = await send_initialize(read, write)
        print("Connected:", init.serverInfo.name)

anyio.run(main)

# TLS (secure transport)
async def main_secure():
    params = HttpClientParameters(
        url="https://mcp.example.com/mcp",
        timeout_s=30,
        headers={"Authorization": "Bearer <token>"}
    )
    async with http_client(params) as (read, write):
        init = await send_initialize(read, write)
        print("Connected:", init.serverInfo.name)

anyio.run(main_secure)

Instalación

Con uv (recomendado)

uv add chuk-mcp                           # core (Python 3.11+ required)
uv add "chuk-mcp[pydantic]"               # add typed Pydantic models (Pydantic v2 only)
uv add "chuk-mcp[http]"                   # add Streamable HTTP transport extras
uv add "chuk-mcp[fast-json]"              # add fast JSON (orjson - 4x faster!)
uv add "chuk-mcp[full]"                   # full install with all features

Con pip

pip install "chuk-mcp"
pip install "chuk-mcp[pydantic]"          # Pydantic v2 only
pip install "chuk-mcp[http]"              # httpx>=0.28 for Streamable HTTP
pip install "chuk-mcp[fast-json]"         # orjson>=3.10 for 4x faster JSON
pip install "chuk-mcp[full]"              # all features

Consejo de rendimiento: Instala [fast-json] para operaciones JSON 4 veces más rápidas (serialización 6.5 veces, deserialización 2.4 veces)

(Requiere pydantic>=2.11.1,<3, httpx>=0.28.1,<1 y orjson>=3.10.0,<4 para extras.)

Versiones de Python: Requiere Python 3.11+; consulta la insignia para las versiones probadas.

Verifica:

python -c "import chuk_mcp; print('✅ chuk-mcp ready')"

Inicio Rápido

Inicialización mínima (servidor demo en línea)

import anyio
from chuk_mcp import StdioServerParameters, stdio_client
from chuk_mcp.protocol.messages import send_initialize

async def main():
    params = StdioServerParameters(
        command="python",
        args=["-c", "import json,sys; req=json.loads(sys.stdin.readline()); print(json.dumps({\"id\":req['id'],\"result\":{\"serverInfo\":{\"name\":\"Demo\",\"version\":\"1.0\"},\"protocolVersion\":\"<negotiated-by-client>\",\"capabilities\":{}}}))"]
    )
    async with stdio_client(params) as (read, write):
        res = await send_initialize(read, write)
        print("Connected:", res.serverInfo.name)

anyio.run(main)

Nota: La versión del protocolo se negocia durante initialize; evita codificarla de forma fija.

Usuarios de Windows: cmd/PowerShell de Windows puede almacenar en búfer stdio de manera diferente. Usa uv run o WSL para desarrollo local si encuentras interbloqueos.

Ejecútalo:

uv run python examples/quickstart_minimal.py

Servidor real (ejemplo SQLite con verificación de capacidades)

import anyio
from chuk_mcp import StdioServerParameters, stdio_client
from chuk_mcp.protocol.messages import send_initialize
from chuk_mcp.protocol.messages.tools import send_tools_call, send_tools_list

async def main():
    params = StdioServerParameters(command="uvx", args=["mcp-server-sqlite", "--db-path", "example.db"])
    async with stdio_client(params) as (read, write):
        # Initialize and check capabilities
        init = await send_initialize(read, write)

        # Capability-gated behavior
        if hasattr(init.capabilities, 'tools'):
            tools = await send_tools_list(read, write)
            print("Tools:", [t.name for t in tools.tools])
            result = await send_tools_call(read, write, name="read_query", arguments={"query": "SELECT 1 as x"})
            print("Result:", result.content)
        else:
            print("Server does not support tools")

anyio.run(main)

Ejecútalo:

# Install SQLite server
uv tool install mcp-server-sqlite

# Run example
uv run python examples/quickstart_sqlite.py

Servidor mínimo (capa de protocolo)

Construye tu propio servidor MCP usando la misma capa de protocolo. Consulta examples/e2e_*_server.py para servidores funcionales completos:

# Conceptual example — for a runnable server, see examples/e2e_*_server.py
import anyio
from chuk_mcp.server import MCPServer, run_stdio_server
from chuk_mcp.protocol.types import ServerCapabilities, ToolCapabilities

async def main():
    server = MCPServer(
        name="demo-server",
        version="0.1.0",
        capabilities=ServerCapabilities(tools=ToolCapabilities())
    )

    # Register handlers using the protocol layer
    async def handle_tools_list(message, session_id):
        # Return (response, notifications). Second value is reserved for
        # optional out-of-band notifications; use None if not sending any.
        return server.protocol_handler.create_response(
            message.id,
            {"tools": [{
                "name": "greet",
                "description": "Say hello",
                "inputSchema": {
                    "type": "object",
                    "properties": {"name": {"type": "string"}},
                    "required": ["name"]
                }
            }]}
        ), None

    server.protocol_handler.register_method("tools/list", handle_tools_list)
    await run_stdio_server(server)

anyio.run(main)

Acompáñalo con un cliente:

# See examples/ for complete client-server pairs
uv run python examples/e2e_tools_client.py

Los ejemplos anteriores usan stdio. Cambia el transporte para hablar con servidores remotos (consulta Transportes).


Conceptos Clave

Herramientas

Descubre y llama funciones expuestas por el servidor.

from chuk_mcp.protocol.messages.tools import send_tools_list, send_tools_call

# list
tools = await send_tools_list(read, write)
for t in tools.tools:
    print(t.name, "-", t.description)

# call
call = await send_tools_call(read, write, name="greet", arguments={"name": "World"})
print(call.content)

Consulta el ejemplo completo: examples/e2e_tools_client.py

Recursos

Lista/lee (y opcionalmente suscríbete a) fuentes de datos.

from chuk_mcp.protocol.messages.resources import send_resources_list, send_resources_read

resources = await send_resources_list(read, write)
if resources.resources:
    uri = resources.resources[0].uri
    data = await send_resources_read(read, write, uri)
    print(data.contents)

Consulta ejemplos completos:

Indicaciones

Plantillas de indicaciones parametrizadas y reutilizables.

from chuk_mcp.protocol.messages.prompts import send_prompts_list, send_prompts_get

prompts = await send_prompts_list(read, write)
if prompts.prompts:
    got = await send_prompts_get(read, write, name=prompts.prompts[0].name, arguments={})
    for m in got.messages:
        print(m.role, m.content)

Consulta el ejemplo completo: examples/e2e_prompts_client.py

Raíces (opcional)

Anuncia directorios que el cliente autoriza al servidor a acceder.

from chuk_mcp.protocol.messages.roots import send_roots_list
roots = await send_roots_list(read, write)  # if supported

Consulta el ejemplo completo: examples/e2e_roots_client.py

Muestreo y Finalización (opcional)

Algunos servidores pueden pedir al cliente que muestree texto o proporcione finalización para argumentos. Estas funciones son opcionales y están controladas por capacidades.

Consulta ejemplos completos:


Transportes

chuk-mcp separa limpiamente el protocolo del transporte, de modo que puedes usar los mismos manejadores de protocolo con cualquier capa de transporte:

  • Stdio — ideal para servidores de procesos secundarios locales
  • HTTP transmisible — habla con servidores remotos sobre HTTP (fragmentado/NDJSON)
  • SSE (Eventos Enviados por el Servidor) — para integraciones de navegador/IDE con empuje unidireccional del servidor
  • Extensible — implementa tu propio transporte adaptando la simple interfaz asíncrona (read, write)

Nota: chuk-mcp es totalmente asíncrono (AnyIO). Usa anyio.run(...) o intégrate en tu bucle de eventos.

Nota: Las capacidades del protocolo se negocian durante initialize, independientemente del transporte. Tú eliges el transporte (stdio o HTTP transmisible) según las necesidades de despliegue/runtime.

Seguridad de hilos: Las instancias de cliente no son seguras entre hilos a través de bucles de eventos. Consulta Preguntas Frecuentes para detalles.

HTTP transmisible usa NDJSON fragmentado. Configura HttpClientParameters(timeout_s=30, headers={"Authorization": "Bearer ..."}). Los clientes transmiten NDJSON con contrapresión. Para cargas útiles grandes, prefiere fragmentos NDJSON sobre blobs base64 para evitar picos de memoria.

Encuadre: HTTP transmisible usa NDJSON (un objeto JSON por línea). Los servidores deben vaciar después de cada objeto; los proxies no deben almacenar en búfer indefinidamente.

Compresión: Habilita gzip en el proxy para reducir flujos de contenido grandes. Las cargas útiles MCP se comprimen bien.

Diseño de la Capa de Protocolo: La capa de protocolo es intencionalmente limpia y mínima: los errores se lanzan inmediatamente sin reintentos. Este diseño mantiene la capa de protocolo enfocada en el transporte de mensajes y el cumplimiento de la especificación MCP. Para casos de uso que requieran lógica de reintentos, manejo de errores, limitación de velocidad o almacenamiento en caché, usa chuk-tool-processor, que proporciona envoltorios componibles para reintentos con retroceso exponencial, limitación de velocidad y almacenamiento en caché. Esta separación de preocupaciones te permite elegir la estrategia de reintentos adecuada para las necesidades específicas de tu aplicación.

Seguridad: Al exponer HTTP transmisible, termina TLS en un proxy y exige autenticación (por ejemplo, tokens de portador). Para CAs privadas, configura el almacén de confianza de tu cliente (por ejemplo, SSL_CERT_FILE=/path/ca.pem, REQUESTS_CA_BUNDLE o SSL_CERT_DIR). La capa de protocolo es agnóstica al transporte y no impone autenticación.


Ejemplos de Configuración

Configuración JSON (el cliente decide cómo generar/conectar)

{
  "mcpServers": {
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "database.db"]
    },
    "filesystem": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-filesystem", "."]
    }
  }
}

Cargando configuración en código

from chuk_mcp import StdioServerParameters, stdio_client
from chuk_mcp.protocol.messages import send_initialize

params = StdioServerParameters(command="uvx", args=["mcp-server-sqlite", "--db-path", "database.db"])
async with stdio_client(params) as (read, write):
    init = await send_initialize(read, write)
    print("Connected to", init.serverInfo.name)

Ejemplos y Demostraciones de Funciones

El directorio examples/ contiene demostraciones completas y funcionales de todas las funciones MCP:

Ejemplos de Inicio Rápido

Ejemplos de Extremo a Extremo (E2E)

Pares completos cliente-servidor construidos con chuk-mcp puro, demostrando tanto la implementación del cliente como del servidor para cada función MCP:

Funciones Principales:

Funciones Avanzadas:

Manejo de Errores:

Ejemplos en ejecución:

Muchos ejemplos E2E son autocontenidos con su propio servidor a nivel de protocolo construido usando chuk-mcp puro. Cuando es relevante, el cliente inicia el servidor de demostración correspondiente:

# Run any example directly - the client will start its server
uv run python examples/e2e_tools_client.py

# Test all E2E examples
for example in examples/e2e_*_client.py; do
    echo "Testing $example"
    uv run python "$example" || exit 1
done

Nota: Cuando es relevante, los ejemplos incluyen un e2e_*_server.py correspondiente que muestra un servidor mínimo construido con la misma capa de protocolo.

Consulte examples/README.md para obtener documentación detallada de todos los ejemplos.


Versionado y Compatibilidad

  • chuk-mcp sigue las revisiones de la especificación MCP y negocia capacidades en initialize.
  • Las funciones más nuevas están restringidas por capacidades y degradan correctamente con servidores más antiguos.
  • La validación/escritura opcional usa Pydantic si está disponible; de lo contrario, un respaldo ligero.

📋 Versiones de Protocolo Compatibles (a partir de v0.1.x)

VersiónEstadoPolítica de Soporte
2025-06-18Más recienteSoporte principal, todas las funciones
2025-03-26EstableCompatibilidad completa, mantenida
2024-11-05HeredadaCompatibilidad hacia atrás, deprecación por decidir

Plataformas probadas: Linux, macOS, Windows (Python 3.11+)

Política de soporte: Las versiones más recientes y estables reciben soporte completo. El soporte de la versión heredada se mantendrá hasta 2026-Q2, después de lo cual podría quedar obsoleta. Consulte el registro de cambios para obtener orientación sobre la migración.

📊 Matriz de Soporte de Funciones del Cliente

Categoría de Función2024-11-052025-03-262025-06-18Estado de Implementación
Operaciones Principales
Herramientas (listar/llamar)✅✅✅✅ Completo
Recursos (listar/leer/suscribirse)✅✅✅✅ Completo
Prompts (listar/obtener)✅✅✅✅ Completo
Transporte
Stdio✅✅✅✅ Completo
HTTP Transmisible–✅✅✅ Completo
Funciones Avanzadas
Muestreo✅✅✅✅ Completo
Finalización✅✅✅✅ Completo
Raíces✅✅✅✅ Completo
Indagación❌❌✅✅ Completo
Funciones de Calidad
Seguimiento de Progreso✅✅✅✅ Completo
Cancelación✅✅✅✅ Completo
Notificaciones✅✅✅✅ Completo
Registro✅✅✅✅ Completo
Anotaciones✅✅✅✅ Completo

Las funciones degradan correctamente al interactuar con servidores más antiguos.

Consulte el registro de cambios para conocer las versiones exactas de la especificación compatibles y cualquier deprecación.

Política de Versionado

Este proyecto sigue Versionado Semántico para las API públicas bajo chuk_mcp.*:

  • Mayor (X.0.0): Cambios que rompen la compatibilidad en las API públicas
  • Menor (0.X.0): Nuevas funciones, compatible hacia atrás
  • Parche (0.0.X): Correcciones de errores, compatible hacia atrás

Cambios que Rompen la Compatibilidad y Migración

v0.7.2: Cambios en el Manejo de Excepciones

Qué Cambió: send_initialize() y send_initialize_with_client_tracking() ahora siempre lanzan excepciones en lugar de devolver None en caso de errores.

Por Qué: Esto permite un manejo de errores adecuado, la reautenticación automática de OAuth en herramientas posteriores (como mcp-cli) y sigue las mejores prácticas de Python.

Guía de Migración:

Antes (v0.7.1 y anteriores):

result = await send_initialize(read, write)
if result is None:
    logging.error("Initialization failed")
    return
# Use result
print(f"Connected to {result.serverInfo.name}")

Después (v0.7.2+):

try:
    result = await send_initialize(read, write)
    # Success - result is guaranteed to be InitializeResult (not None)
    print(f"Connected to {result.serverInfo.name}")
except RetryableError as e:
    # Handle retryable errors (e.g., 401 authentication)
    logging.error(f"Retryable error: {e}")
except VersionMismatchError as e:
    # Handle version incompatibility
    logging.error(f"Version mismatch: {e}")
except TimeoutError as e:
    # Handle timeout
    logging.error(f"Timeout: {e}")
except Exception as e:
    # Handle other errors
    logging.error(f"Error: {e}")

Cambios en el Tipo de Retorno:

  • send_initialize(): Optional[InitializeResult] → InitializeResult
  • send_initialize_with_client_tracking(): Optional[InitializeResult] → InitializeResult

Beneficios:

  • ✅ Reautenticación automática de OAuth en mcp-cli
  • ✅ Propagación de errores y depuración adecuadas
  • ✅ Seguridad de tipos (no se necesitan comprobaciones de Optional)
  • ✅ Contexto completo de excepciones con seguimientos de pila

Ver También:


Comparación con el SDK Oficial de MCP

Funciónchuk-mcpSDK Oficial de MCP para Python
FilosofíaBiblioteca de cumplimiento de protocoloMarco completo
AlcanceCliente + Servidor, centrado en protocoloMarco de Cliente + Servidor
EscrituraPydantic opcional (respaldo disponible)Pydantic requerido
Transportesstdio, SSE, HTTP Transmisible (conectable)stdio, SSE, HTTP Transmisible
Navegador/WASMCompatible con PyodideVaría / no es un objetivo principal
DependenciasMínimas (núcleo anyio)Pila más pesada
Marco de ServidorAsistentes ligerosEstructura de servidor opinada
Estilo de APIFunciones explícitas send_*Abstracciones de nivel superior
Caso de Uso ObjetivoIntegración de protocolo, clientes/servidores personalizadosAplicaciones MCP completas
OrquestaciónExterna (usted elige)Patrones integrados
Curva de AprendizajeBaja (nivel de protocolo)Media (conceptos de marco)

Cuándo elegir chuk-mcp:

  • Construir clientes o servidores MCP personalizados
  • Necesitar flexibilidad de transporte (HTTP Transmisible)
  • Querer dependencias mínimas
  • Preferir control a nivel de protocolo
  • Ejecutar en entornos restringidos (WASM, funciones de borde)
  • Necesitar integrar MCP en aplicaciones existentes

Ejemplo del mundo real: chuk-mcp-server usa chuk-mcp como su capa de cumplimiento de protocolo

Cuándo elegir el SDK oficial:

  • Construir servidores MCP completos rápidamente con patrones opinados
  • Querer abstracciones de marco listas para usar
  • Usar principalmente transporte stdio
  • Preferir API de nivel superior

Objetivos de Diseño y No Objetivos

Objetivos

  • Ser la forma más simple de implementar MCP en Python (cliente o servidor)
  • Mantener la API pequeña, explícita y tipada
  • Hacer que los transportes sean conectables y la lógica de protocolo reutilizable
  • Soportar tanto casos de uso de cliente como de servidor con abstracciones ligeras

No Objetivos

  • Competir con marcos de agentes completos / IDEs
  • Incorporar estructura de aplicación opinada o motores de flujo de trabajo
  • Incluir dependencias pesadas por defecto
  • Proporcionar orquestación de alto nivel (esa es su capa de aplicación)

Escalado y Concurrencia

chuk-mcp maneja cientos de conexiones concurrentes de manera eficiente con un uso mínimo de recursos:

Puntos de Referencia de Concurrencia

Rendimiento Probado (consulte benchmarks/PERFORMANCE_REPORT.md para más detalles):

  • Más de 700 conexiones concurrentes probadas con éxito (se detuvo por tiempo de espera, no por límite de capacidad)
  • Más de 252 conexiones/segundo de rendimiento para cambios rápidos de conexión
  • ~34KB de memoria por conexión con escalado lineal
  • Cero fugas de memoria verificado en más de 200 iteraciones

Estimaciones de Capacidad:

  • Escala pequeña (< 100 agentes): 512MB de RAM, 1 núcleo
  • Escala media (100-1,000 agentes): 1-2GB de RAM, 2-4 núcleos
  • Escala grande (1,000-10,000 agentes): 4-8GB de RAM, 8+ núcleos
  • Escala empresarial (10,000+ agentes): Se recomienda equilibrio de carga

Mejores Prácticas

Patrón: Crear todos → Inicializar todos (Secuencial)

# RECOMMENDED: Fastest pattern for multiple agents
agent1 = create_agent(mcp_config1)
agent2 = create_agent(mcp_config2)
agent3 = create_agent(mcp_config3)

# Then initialize
await agent1.initialize_tools()
await agent2.initialize_tools()
await agent3.initialize_tools()

Patrón: Intercalado (También Compatible)

# WORKS: Fixed in v0.8.1 with lazy stream initialization
agent1 = create_agent(mcp_config1)
await agent1.initialize_tools()

agent2 = create_agent(mcp_config2)
await agent2.initialize_tools()

agent3 = create_agent(mcp_config3)  # No longer hangs!
await agent3.initialize_tools()

Importante: Use siempre StdioClient como administrador de contexto asíncrono:

# CORRECT: Streams initialized in async context
async with StdioClient(params) as client:
    # Use client here
    pass

# INCORRECT: Don't access streams before __aenter__
client = StdioClient(params)
client.get_streams()  # ❌ Raises RuntimeError

Recomendaciones de Monitoreo

Para implementaciones, monitoree estas métricas:

  • Conexiones Activas: Realice un seguimiento del número de clientes concurrentes
  • Crecimiento de Memoria: Debe permanecer plano con el tiempo (~0.034MB por conexión)
  • Descriptores de Archivo: Monitoree mediante lsof o /proc/<pid>/fd
  • Tasa de Éxito de Conexión: Debe mantenerse al 100%

Consulte benchmarks/PERFORMANCE_REPORT.md para obtener un análisis de rendimiento detallado y pautas de implementación.


Preguntas Frecuentes

P: ¿Esto incluye un marco de servidor?

R: Sí, en la capa de protocolo. chuk-mcp proporciona mensajes tipados y asistentes utilizables tanto en clientes como en servidores, pero no es un marco de servidor opinado—usted aporta su propia estructura/orquestación de aplicación.

P: ¿Se requiere Pydantic?

R: No. Si está instalado (solo Pydantic v2), obtendrá tipos y validación más ricos. Si no, la biblioteca usa un respaldo ligero con modelos basados en diccionarios.

P: ¿Qué transporte debería usar?

R: Use stdio para desarrollo local y procesos secundarios. Use HTTP Transmisible para servidores remotos detrás de TLS con autenticación.

P: ¿Dónde puedo encontrar más ejemplos?

R: Consulte el directorio examples/ para ver demostraciones completas de todas las funciones de MCP, incluidos ejemplos de inicio rápido y pares completos de cliente-servidor de extremo a extremo. Para una implementación de servidor del mundo real, consulte chuk-mcp-server que usa chuk-mcp como su biblioteca de protocolo.

P: ¿Cómo pruebo mi implementación?

R: Ejecute make test o uv run pytest para ejecutar la suite de pruebas. Use make examples (si está presente) para probar todos los ejemplos E2E. Consulte la sección Contribuyendo para más detalles.

P: ¿Está listo para usar?

R: Sí. chuk-mcp está implementado a escala. Incluye manejo de errores, seguridad de tipos y sigue las especificaciones del protocolo MCP. Consulte los informes de cobertura de pruebas para obtener métricas de confianza.

P: ¿Es seguro para subprocesos?

R: Las instancias de cliente no son seguras para subprocesos entre bucles de eventos. Comparta un cliente dentro de un único bucle asíncrono; use instancias separadas por bucle/subproceso.

P: ¿Qué no está incluido?

R: Autenticación, terminación TLS, persistencia y orquestación son preocupaciones de la aplicación—apórtelas usted mismo. chuk-mcp proporciona solo cumplimiento de protocolo. Para frontends de navegador/WASM con CORS y TLS, termine TLS en el proxy y establezca Access-Control-Allow-Origin en el origen de su frontend; evite * con credenciales.

P: ¿Cómo agrego lógica de reintento y limitación de velocidad?

R: Use chuk-tool-processor que proporciona envoltorios componibles para reintentos (con retroceso exponencial), limitación de velocidad y almacenamiento en caché. chuk-mcp se centra en el cumplimiento del protocolo; chuk-tool-processor maneja las preocupaciones de ejecución.

P: ¿Cuáles son los errores comunes y cómo los manejo?

R: Excepciones comunes y acciones recomendadas:

Tipo de ErrorCódigo JSON-RPCAcción
Error de análisis-32700Corrija la sintaxis JSON en la solicitud
Solicitud no válida-32600Verifique los campos requeridos (jsonrpc, method, id)
Método no encontrado-32601Verifique el nombre del método y las capacidades del servidor
Parámetros no válidos-32602Valide los tipos de parámetros y los campos requeridos
Error interno-32603Revise los registros del servidor, reintente la operación
Error de autenticación (401)-32603Reautentique (automático en mcp-cli)
Solicitud cancelada-32800Maneje la cancelación correctamente
Contenido demasiado grande-32801Reduzca el tamaño de la carga útil o use transmisión
Conexión/TransportevaríaVerifique la red, confirme que el servidor esté en ejecución

Nota: Al usar transportes basados en HTTP (SSE o HTTP Transmisible), los errores de la capa de transporte (fallos de red, problemas TLS, problemas de autenticación) aparecerán como códigos de estado HTTP antes de llegar a la capa de protocolo MCP. Sin embargo, una vez establecido el transporte, todos los errores de protocolo MCP siguen el sistema de códigos de error JSON-RPC mostrado arriba.

Todos los errores de protocolo heredan de clases de excepción base y siempre se lanzan (nunca devuelven None). Consulte los ejemplos para ver los patrones de manejo de errores.

Mejores Prácticas de Manejo de Excepciones:

from chuk_mcp.protocol.types.errors import (
    RetryableError,
    NonRetryableError,
    VersionMismatchError
)
from chuk_mcp.protocol.messages import send_initialize

try:
    # Initialize connection
    result = await send_initialize(read, write)
    # Success - result is guaranteed to be InitializeResult (not None)
    print(f"Connected to {result.serverInfo.name}")

except VersionMismatchError as e:
    # Protocol version incompatibility - cannot recover
    logging.error(f"Version mismatch: {e}")
    # Disconnect and inform user

except RetryableError as e:
    # Retryable errors (e.g., 401 authentication failures)
    if "401" in str(e).lower() or "unauthorized" in str(e).lower():
        # Trigger OAuth re-authentication
        # In mcp-cli, this happens automatically
        logging.info("Re-authenticating...")
    else:
        # Other retryable errors - implement retry logic
        logging.warning(f"Retryable error: {e}")

except TimeoutError as e:
    # Server didn't respond in time
    logging.error(f"Timeout: {e}")
    # Retry with longer timeout or check server status

except NonRetryableError as e:
    # Non-retryable errors - log and fail
    logging.error(f"Fatal error: {e}")

except Exception as e:
    # Other unexpected errors
    logging.error(f"Unexpected error: {e}")

Consulte examples/initialize_error_handling.py para ver demostraciones completas de manejo de errores.


Contribuyendo

¡Se aceptan PRs! Por favor:

  1. Abra primero un problema pequeño y enfocado (opcional pero útil).
  2. Agregue pruebas y sugerencias de tipos para nuevas funcionalidades.
  3. Mantenga las API públicas mínimas y consistentes.
  4. Ejecute los linters y la suite de pruebas antes de enviar.

Los PRs deben mantener una cobertura ≥85%; se aplica en CI junto con comprobaciones de tipos mypy y linting ruff.

# Clone and setup
git clone <repository-url>
# or install from PyPI: pip install chuk-mcp
cd chuk-mcp
uv sync

# Install pre-commit hooks (optional)
pre-commit install

# Run examples
uv run python examples/quickstart_minimal.py

# Run tests
uv run pytest

# Type checking
uv run mypy src/chuk_mcp

# Or use the Makefile (if present)
make test
make typecheck
make lint
make examples

Informes de errores / solicitudes de funciones: Plantillas de problemas disponibles en .github/

Código de Conducta: Se espera que los contribuyentes sigan el Pacto del Contribuyente

Seguridad

Si cree que ha encontrado un problema de seguridad, repórtelo abriendo un aviso de seguridad en el repositorio de GitHub en lugar de abrir un problema público.


Exhibición de Funciones

Esta sección proporciona fragmentos de código detallados que demuestran las características de MCP. Todos los ejemplos incluyen seguridad de tipos completa.

🔧 Herramientas — Llamada de Funciones

Las herramientas son funciones que la IA puede invocar:

from chuk_mcp.protocol.messages.tools import send_tools_list, send_tools_call
from chuk_mcp.protocol.types.content import parse_content, TextContent

# List all available tools — returns typed ListToolsResult
tools_result = await send_tools_list(read, write)
print(f"📋 Available tools: {len(tools_result.tools)}")

for tool in tools_result.tools:
    print(f"  • {tool.name}: {tool.description}")

# Call a tool — returns typed ToolResult
result = await send_tools_call(
    read, write,
    name="greet",
    arguments={"name": "World"}
)

# Parse content with type safety
content = parse_content(result.content[0])
assert isinstance(content, TextContent)
print(f"✅ Result: {content.text}")

Ejemplo completo: uv run python examples/e2e_tools_client.py

📄 Recursos — Lectura de Datos

Los recursos proporcionan acceso a fuentes de datos (archivos, bases de datos, APIs):

from chuk_mcp.protocol.messages.resources import send_resources_list, send_resources_read

# List available resources — returns typed ListResourcesResult
resources_result = await send_resources_list(read, write)
print(f"📚 Found {len(resources_result.resources)} resources")

for resource in resources_result.resources:
    print(f"  • {resource.name}")
    print(f"    URI: {resource.uri}")

# Read a resource — returns typed ReadResourceResult
if resources_result.resources:
    uri = resources_result.resources[0].uri
    read_result = await send_resources_read(read, write, uri)

    for content in read_result.contents:
        if hasattr(content, 'text'):
            print(f"📖 Content: {content.text[:200]}...")

Ejemplo completo: uv run python examples/e2e_resources_client.py

📡 Suscripciones a Recursos — Actualizaciones en Vivo

Suscríbete a recursos para recibir notificaciones de cambios en tiempo real:

from chuk_mcp.protocol.messages.resources import (
    send_resources_subscribe,
    send_resources_unsubscribe
)

# Subscribe to a resource
uri = "file:///logs/app.log"
success = await send_resources_subscribe(read, write, uri)

if success:
    print(f"✅ Subscribed to {uri}")
    print("📡 Listening for changes...")

    # In a real app, handle notifications in a loop
    # Notifications arrive as messages from the server

    # Unsubscribe when done
    await send_resources_unsubscribe(read, write, uri)
    print("🔕 Unsubscribed")

Ejemplo completo: uv run python examples/e2e_subscriptions_client.py

💬 Prompts — Gestión de Plantillas

Los prompts son plantillas reutilizables con parámetros:

from chuk_mcp.protocol.messages.prompts import send_prompts_list, send_prompts_get

# List available prompts — returns typed ListPromptsResult
prompts_result = await send_prompts_list(read, write)
print(f"💬 Available prompts: {len(prompts_result.prompts)}")

for prompt in prompts_result.prompts:
    print(f"  • {prompt.name}: {prompt.description}")
    if hasattr(prompt, 'arguments') and prompt.arguments:
        args = [a.name for a in prompt.arguments]
        print(f"    Arguments: {', '.join(args)}")

# Get a prompt with arguments — returns typed GetPromptResult
prompt_result = await send_prompts_get(
    read, write,
    name="code_review",
    arguments={"file": "main.py", "language": "python"}
)

# Use the formatted messages
for message in prompt_result.messages:
    print(f"🤖 {message.role}: {message.content}")

Ejemplo completo: uv run python examples/e2e_prompts_client.py

🎯 Muestreo — Generación de Contenido con IA

Permite que los servidores soliciten a la IA generar contenido en su nombre (requiere aprobación del usuario):

from chuk_mcp.protocol.messages.sampling import sample_text

# Check if server supports sampling
if hasattr(init_result.capabilities, 'sampling'):
    print("✅ Server supports sampling")

    # Server requests AI to generate content using helper
    result = await sample_text(
        read, write,
        prompt="Explain quantum computing in simple terms",
        max_tokens=1000,
        model_hint="claude",
        temperature=0.7
    )

    # Access typed response
    if hasattr(result.content, 'text'):
        print(f"🤖 AI Generated: {result.content.text}")

    print(f"📊 Model: {result.model}")
    print(f"🔢 Stop Reason: {result.stopReason or 'N/A'}")

Caso de uso: Los servidores pueden usar el muestreo para generar código, documentación o análisis basados en los datos a los que tienen acceso.

Ejemplo completo: uv run python examples/e2e_sampling_client.py

📁 Raíces — Control de Acceso a Directorios

Las raíces definen qué directorios permite el cliente que los servidores accedan.

from chuk_mcp.protocol.messages.roots import (
    send_roots_list,
    send_roots_list_changed_notification
)

# Check if server supports roots
if hasattr(init_result.capabilities, 'roots'):
    print("✅ Server supports roots capability")

    # List current roots — returns typed ListRootsResult
    roots_result = await send_roots_list(read, write)

    print(f"📁 Available roots: {len(roots_result.roots)}")
    for root in roots_result.roots:
        print(f"  • {root.name}: {root.uri}")

    # Notify server when roots change
    await send_roots_list_changed_notification(write)
    print("📢 Notified server of roots change")

Caso de uso: Controla qué directorios puede acceder la IA, permitiendo operaciones seguras en entornos aislados.

Ejemplo completo: uv run python examples/e2e_roots_client.py

🎭 Elicitación — Solicitudes de Entrada del Usuario

La elicitación permite que los servidores soliciten entrada estructurada de los usuarios:

from chuk_mcp.protocol.messages.elicitation import send_elicitation_request

# Server requests user input
response = await send_elicitation_request(
    read, write,
    prompt="Enter API credentials",
    fields=[
        {"name": "api_key", "type": "text", "required": True},
        {"name": "region", "type": "select", "options": ["us", "eu", "asia"]}
    ]
)

# Access user's input
print(f"User provided: {response.values}")

Caso de uso: Flujos de trabajo interactivos, flujos OAuth, diálogos de confirmación.

Ejemplo completo: uv run python examples/e2e_elicitation_client.py

💡 Completado — Autocompletado Inteligente

Obtén sugerencias inteligentes para argumentos de herramientas:

from chuk_mcp.protocol.messages.completions import (
    send_completion_complete,
    create_argument_info
)

# Get completions for a file path argument — returns typed CompletionResult
response = await send_completion_complete(
    read, write,
    ref={"type": "ref/resource", "uri": "file:///data/"},
    argument=create_argument_info(
        name="filename",
        value="sales_202"  # Partial input
    )
)

# Show suggestions
print("💡 Suggestions for 'sales_202':")
for value in response.completion.values:
    print(f"  • {value}")

Ejemplo completo: uv run python examples/e2e_completion_client.py

📊 Seguimiento de Progreso

Monitorea operaciones de larga duración con actualizaciones de progreso:

from chuk_mcp.protocol.messages.tools import send_tools_call

# Call a long-running tool
# Progress notifications will be sent automatically
print("🔄 Starting long operation...")

result = await send_tools_call(
    read, write,
    name="process_large_dataset",
    arguments={"dataset": "sales_data.csv"}
)

print("✅ Operation complete")
# Progress notifications are handled automatically by the client

Ejemplo completo: uv run python examples/e2e_progress_client.py

🚫 Cancelación

Cancela operaciones de larga duración con tiempo de espera:

import anyio
from chuk_mcp.protocol.messages.cancellation import send_cancelled_notification
from chuk_mcp.protocol.messages.tools import send_tools_call

async def cancel_after_timeout():
    request_id = "long-op-123"

    async with anyio.create_task_group() as tg:
        # Start long-running operation
        tg.start_soon(send_tools_call, read, write, "process_large_dataset",
                      {"dataset": "big.csv"}, request_id)

        # Cancel after 2 seconds
        with anyio.move_on_after(2):
            await anyio.sleep(999)

        # Send cancellation
        await send_cancelled_notification(write, request_id=request_id, reason="timeout")
        print("🚫 Cancellation sent")

anyio.run(cancel_after_timeout)

Ejemplo completo: uv run python examples/e2e_cancellation_client.py

🌐 Múltiples Transportes

Usa diferentes protocolos de transporte para diferentes escenarios:

import anyio
from chuk_mcp.protocol.messages import send_initialize
from chuk_mcp import stdio_client, StdioServerParameters
from chuk_mcp.transports.http import http_client, HttpClientParameters

async def main():
    # Stdio transport (local processes)
    p1 = StdioServerParameters(
        command="uvx",
        args=["mcp-server-sqlite", "--db-path", "local.db"]
    )
    async with stdio_client(p1) as (r, w):
        init = await send_initialize(r, w)
        print("📡 Stdio:", init.serverInfo.name)

    # Streamable HTTP transport (remote servers)
    p2 = HttpClientParameters(url="http://localhost:8989/mcp")
    async with http_client(p2) as (r, w):
        init = await send_initialize(r, w)
        print("🌐 Streamable HTTP:", init.serverInfo.name)

anyio.run(main)

🔄 Orquestación Multi-Servidor

Conéctate a múltiples servidores simultáneamente:

from chuk_mcp import stdio_client, StdioServerParameters
from chuk_mcp.protocol.messages import send_initialize
from chuk_mcp.protocol.messages.tools import send_tools_list

servers = [
    StdioServerParameters(
        command="uvx",
        args=["mcp-server-sqlite", "--db-path", "db1.db"]
    ),
    StdioServerParameters(
        command="npx",
        args=["-y", "@modelcontextprotocol/server-filesystem", "."]
    )
]

print("🔗 Connecting to multiple servers...")

for i, server_params in enumerate(servers, 1):
    try:
        async with stdio_client(server_params) as (read, write):
            init_result = await send_initialize(read, write)
            tools_result = await send_tools_list(read, write)

            print(f"\n📡 Server {i}: {init_result.serverInfo.name}")
            print(f"   Tools: {len(tools_result.tools)}")

            # Show first 3 tools
            for tool in tools_result.tools[:3]:
                print(f"   • {tool.name}")
    except Exception as e:
        print(f"⚠️ Server {i} failed: {e}")

Seguridad de Tipos y Validación

Todos los mensajes del protocolo devuelven resultados completamente tipados usando Pydantic (o validación de respaldo):

from chuk_mcp.protocol.types.content import parse_content, TextContent
from chuk_mcp.protocol.messages.tools import send_tools_call

# Call a tool and get a typed result
tool_result = await send_tools_call(read, write, name="greet", arguments={"name": "World"})

# Type-safe content parsing
content = parse_content(tool_result.content[0])
assert isinstance(content, TextContent)
print(content.text)

Beneficios:

  • Retornos tipados: Todas las funciones de send_* devuelven modelos Pydantic tipados
  • Análisis de contenido: Usa parse_content() para manejo de contenido con seguridad de tipos
  • Validación en tiempo de ejecución: Validación automática con mensajes de error claros
  • Soporte de IDE: Autocompletado completo y verificación de tipos

Monitoreo y Registro

Características integradas para entornos desplegados:

from chuk_mcp.protocol.messages.logging import send_logging_set_level

# Set server logging level
await send_logging_set_level(write, level="debug")

Características:

  • Registro estructurado con niveles configurables
  • Monitoreo de rendimiento (latencia, tasas de error, rendimiento)
  • Soporte de seguimiento de progreso y cancelación
  • Propagación de errores limpia (sin reintentos automáticos en la capa de protocolo)

Ejemplo completo: uv run python examples/e2e_logging_client.py


Ecosistema

chuk-mcp es parte de un conjunto modular de herramientas MCP de Python:

  • chuk-tool-processor — Ejecución confiable de llamadas a herramientas con reintentos, caché y retroceso exponencial
  • chuk-mcp-server — Implementación de servidor MCP del mundo real construida sobre chuk-mcp
  • chuk-mcp-cli — CLI interactivo y área de pruebas para probar servidores MCP

Cada componente se enfoca en hacer una cosa bien y puede usarse de forma independiente o conjunta. Todos estos se basan en la capa de protocolo de chuk-mcp, por lo que heredan las mismas características de baja latencia y sobrecarga mínima.


Licencia

Apache 2.0 — ver LICENCIA.