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
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:
| Capa | Propósito | Uso |
|---|---|---|
| CLI y Demo | Utilidades integradas y demostraciones | Opcional: usa la capa de protocolo directamente |
| API de Cliente/Servidor | Abstracciones de alto nivel para interacciones cliente-servidor | Opcional: puede usar la capa de protocolo directamente |
| Capa de Protocolo | Definiciones de mensajes, manejo tipado de solicitudes/respuestas, negociación de capacidades | Núcleo: implementa la especificación MCP |
| Capa de Transporte | Implementaciones de transporte conectables (stdio, HTTP transmisible) | Elige según el despliegue |
| Capa Base | Respaldo de Pydantic, configuración compartida, adaptadores de tipos | Fundació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?
- Rendimiento del Protocolo
- De un Vistazo
- Instalación
- Inicio Rápido
- Conceptos Clave
- Transportes
- Ejemplos de Configuración
- Ejemplos y Demostraciones de Funciones
- Versionado y Compatibilidad
- Comparación con el SDK Oficial de MCP
- Objetivos de Diseño y No Objetivos
- Escalado y Concurrencia
- Preguntas Frecuentes
- Contribuciones
- Vitrina de Funciones
- Ecosistema
- Licencia
¿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,<1yorjson>=3.10.0,<4para 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 runo 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_BUNDLEoSSL_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
quickstart_minimal.py— Configuración mínima de cliente MCPquickstart_sqlite.py— Trabajando con servidor MCP SQLitequickstart_resources.py— Accediendo a recursos del servidorquickstart_complete.py— Demo de múltiples funciones
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:
e2e_tools_client.py— Registro, descubrimiento e invocación de herramientase2e_resources_client.py— Listado y lectura de recursose2e_prompts_client.py— Plantillas de indicaciones reutilizables
Funciones Avanzadas:
e2e_roots_client.py— Gestión de raíces del sistema de archivose2e_sampling_client.py— Solicitudes LLM iniciadas por el servidore2e_completion_client.py— Funcionalidad de autocompletadoe2e_subscriptions_client.py— Notificaciones de cambios de recursose2e_cancellation_client.py— Cancelación de operacionese2e_progress_client.py— Seguimiento de progresoe2e_logging_client.py— Manejo de mensajes de registroe2e_elicitation_client.py— Solicitudes de entrada de usuarioe2e_annotations_client.py— Metadatos de contenido
Manejo de Errores:
initialize_error_handling.py— Patrones integrales de manejo de errores (OAuth 401, desajuste de versión, tiempo de espera, etc.)
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.pycorrespondiente 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-mcpsigue 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ón | Estado | Política de Soporte |
|---|---|---|
2025-06-18 | Más reciente | Soporte principal, todas las funciones |
2025-03-26 | Estable | Compatibilidad completa, mantenida |
2024-11-05 | Heredada | Compatibilidad 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ón | 2024-11-05 | 2025-03-26 | 2025-06-18 | Estado 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]→InitializeResultsend_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:
EXCEPTION_HANDLING_FIX.md- Documentación técnica detalladaexamples/initialize_error_handling.py- Ejemplos completos de manejo de errorestests/mcp/messages/test_initialize_exceptions.py- Suite de pruebas
Comparación con el SDK Oficial de MCP
| Función | chuk-mcp | SDK Oficial de MCP para Python |
|---|---|---|
| Filosofía | Biblioteca de cumplimiento de protocolo | Marco completo |
| Alcance | Cliente + Servidor, centrado en protocolo | Marco de Cliente + Servidor |
| Escritura | Pydantic opcional (respaldo disponible) | Pydantic requerido |
| Transportes | stdio, SSE, HTTP Transmisible (conectable) | stdio, SSE, HTTP Transmisible |
| Navegador/WASM | Compatible con Pyodide | Varía / no es un objetivo principal |
| Dependencias | Mínimas (núcleo anyio) | Pila más pesada |
| Marco de Servidor | Asistentes ligeros | Estructura de servidor opinada |
| Estilo de API | Funciones explícitas send_* | Abstracciones de nivel superior |
| Caso de Uso Objetivo | Integración de protocolo, clientes/servidores personalizados | Aplicaciones MCP completas |
| Orquestación | Externa (usted elige) | Patrones integrados |
| Curva de Aprendizaje | Baja (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
lsofo/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 Error | Código JSON-RPC | Acción |
|---|---|---|
| Error de análisis | -32700 | Corrija la sintaxis JSON en la solicitud |
| Solicitud no válida | -32600 | Verifique los campos requeridos (jsonrpc, method, id) |
| Método no encontrado | -32601 | Verifique el nombre del método y las capacidades del servidor |
| Parámetros no válidos | -32602 | Valide los tipos de parámetros y los campos requeridos |
| Error interno | -32603 | Revise los registros del servidor, reintente la operación |
| Error de autenticación (401) | -32603 | Reautentique (automático en mcp-cli) |
| Solicitud cancelada | -32800 | Maneje la cancelación correctamente |
| Contenido demasiado grande | -32801 | Reduzca el tamaño de la carga útil o use transmisión |
| Conexión/Transporte | varía | Verifique 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:
- Abra primero un problema pequeño y enfocado (opcional pero útil).
- Agregue pruebas y sugerencias de tipos para nuevas funcionalidades.
- Mantenga las API públicas mínimas y consistentes.
- 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.