chuk-mcp
Um cliente Python para o Model Context Protocol (MCP), um padrão aberto para conectar assistentes de IA a dados e ferramentas externas.
Documentação
chuk-mcp
Uma implementação Python enxuta e minimalista do Model Context Protocol (MCP).
Oferece suporte de primeira classe ao protocolo MCP em Python — leve, assíncrono e fiel à especificação desde o primeiro dia.
Requer Python 3.11+
chuk-mcp oferece uma implementação limpa, tipada e agnóstica de transporte para clientes e servidores MCP. Ela foca na superfície do protocolo (mensagens, tipos, versionamento, transportes) e deixa orquestração, UIs e frameworks de agentes para outras camadas.
✳️ O que é: uma biblioteca de conformidade com o protocolo com auxiliares ergonômicos para clientes e servidores.
⛔ O que não é: um runtime de chatbot, mecanismo de fluxo de trabalho ou um framework de aplicação opinativo.
Arquitetura: Onde o chuk-mcp se Encaixa
Visão Geral da Pilha
┌──────────────────────────────────────┐
│ 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) │
└──────────────────────────────────────┘
O chuk-mcp fornece a camada de protocolo — conecte aplicações de IA a ferramentas e fontes de dados usando o protocolo MCP padrão.
Arquitetura Interna
A biblioteca em si é organizada em camadas que você pode usar em diferentes níveis de abstração:
┌─────────────────────────────────────────┐
│ 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
└─────────────────────────────────────────┘
Detalhes das Camadas:
| Camada | Propósito | Uso |
|---|---|---|
| CLI e Demonstração | Utilitários integrados e demonstrações | Opcional — use a camada de protocolo diretamente |
| API de Cliente/Servidor | Abstrações de alto nível para interações cliente-servidor | Opcional — pode usar a camada de protocolo diretamente |
| Camada de Protocolo | Definições de mensagens, tratamento de requisições/respostas com segurança de tipos, negociação de capacidades | Núcleo — implementa a especificação MCP |
| Camada de Transporte | Implementações de transporte plugáveis (stdio, HTTP Streamable) | Escolha com base na implantação |
| Camada Base | Fallback Pydantic, configuração compartilhada, adaptadores de tipo | Fundação — automática |
A maioria dos usuários trabalha com a Camada de Protocolo (funções send_*) e a Camada de Transporte (clientes stdio/HTTP), usando opcionalmente a API de Cliente/Servidor para abstrações de nível mais alto.
Sumário
- Por que chuk‑mcp?
- Desempenho do Protocolo
- De Relance
- Instalação
- Início Rápido
- Conceitos Principais
- Transportes
- Exemplos de Configuração
- Exemplos e Demonstrações de Recursos
- Versionamento e Compatibilidade
- Comparação com o SDK MCP Oficial
- Objetivos de Design e Não‑Objetivos
- Escalabilidade e Concorrência
- FAQ
- Contribuindo
- Vitrine de Recursos
- Ecossistema
- Licença
Por que chuk-mcp?
- Protocolo em primeiro lugar: Foca em mensagens MCP, tipos e negociação de capacidades — spec.modelcontextprotocol.io
- Cliente + Servidor: Suporte completo para construir tanto clientes quanto servidores MCP
- Tipado: Anotações de tipo completas; modelos Pydantic opcionais quando disponíveis
- Agnóstico de transporte: stdio por padrão, HTTP Streamable (NDJSON) para servidores remotos, facilmente extensível
- Assíncrono em primeiro lugar: Construído sobre AnyIO; integre com
anyio.run(...)ou seu loop existente - Pequeno e focado: Sem orquestração pesada ou suposições de agente
- Camada de protocolo limpa: Erros falham rapidamente sem tentativas — traga sua própria estratégia de tratamento de erros
- Confiável: Erros claros, ganchos de registro estruturados, componível com camadas de tentativa/cache
- ⚡ Alto desempenho: Sobrecarga do protocolo na faixa de 2-5ms; JSON rápido opcional para serialização 4x mais rápida. Veja Desempenho do Protocolo para benchmarks detalhados
Desempenho do Protocolo
chuk-mcp é projetado para manter a sobrecarga do protocolo MCP na faixa de 2-5 ms, para que o custo de usar ferramentas seja dominado pelas próprias ferramentas, não pelo protocolo.
Por que é rápido:
- Zero dependências pesadas (apenas núcleo AnyIO)
- stdio nativo assíncrono e HTTP NDJSON
- Sem execução de ferramentas dentro da biblioteca
- Caminho rápido orjson opcional (
[fast-json])
💡 Para números de concorrência e capacidade, veja Escalabilidade e Concorrência.
⚡ Benchmarks de Latência
Sobrecarga do protocolo (medições típicas em hardware moderno):
- Inicializar → Lista de Ferramentas: 2-3 ms
- Chamada de Ferramenta (ida e volta): < 5 ms de sobrecarga (além do tempo real de execução da ferramenta)
- Streaming: Sobrecarga quase zero devido aos limites de chunks NDJSON
Benchmarks executados em macOS (Darwin 24.6.0), Python 3.11 — veja benchmarks/PERFORMANCE_REPORT.md para o ambiente exato e comandos.
🚀 Serialização JSON (Caminho Rápido Opcional)
Instale com [fast-json] para operações JSON ~4x mais rápidas usando orjson:
- Serialização: ~6x mais rápida
- Desserialização: ~2x mais rápida
- Ida e volta: ~4x mais rápida
pip install "chuk-mcp[fast-json]" # Automatic with graceful fallback
Números de benchmark de benchmarks/json_performance.py comparando orjson vs stdlib json em mensagens MCP realistas.
🎯 Casos de Uso Ideais
Isso torna o chuk-mcp perfeito para:
- Chamadas de ferramentas de alta frequência — sobrecarga mínima por requisição
- Agentes em tempo real — latência de protocolo abaixo de 5ms
- UIs de streaming — sobrecarga de chunks NDJSON quase zero
- Processadores de ferramentas — rápido o suficiente para ser transparente
- Ambientes WASM/edge — pegada mínima
- Cargas de trabalho de alto throughput — comprovado em escala (veja Escalabilidade e Concorrência)
De Relance
Experimente agora:
# Install an example MCP server
uv tool install mcp-server-sqlite
# Run the quick-start example
uv run python examples/quickstart_sqlite.py
Hello World
Um servidor MCP mínimo funcional em ~10 linhas:
# 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)
Execute: uv run python hello_mcp.py — ou conecte qualquer cliente MCP via stdio!
Stdio (processos locais):
# 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 Streamable (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)
Instalação
Com 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
Com 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
Dica de desempenho: Instale
[fast-json]para operações JSON 4x mais rápidas (serialização 6.5x, desserialização 2.4x)(Requer
pydantic>=2.11.1,<3,httpx>=0.28.1,<1eorjson>=3.10.0,<4para extras.)
Versões de Python: Requer Python 3.11+; veja o selo para versões testadas.
Verifique:
python -c "import chuk_mcp; print('✅ chuk-mcp ready')"
Início Rápido
Inicialização mínima (servidor de demonstração inline)
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: A versão do protocolo é negociada durante
initialize; evite codificar valores fixos.
Usuários de Windows: O cmd/PowerShell do Windows pode armazenar stdio em buffer de forma diferente. Use
uv runou WSL para desenvolvimento local se encontrar deadlocks.
Execute:
uv run python examples/quickstart_minimal.py
Servidor real (exemplo SQLite com verificação de capacidade)
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)
Execute:
# Install SQLite server
uv tool install mcp-server-sqlite
# Run example
uv run python examples/quickstart_sqlite.py
Servidor mínimo (camada de protocolo)
Construa seu próprio servidor MCP usando a mesma camada de protocolo. Veja examples/e2e_*_server.py para servidores funcionais 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)
Combine com um cliente:
# See examples/ for complete client-server pairs
uv run python examples/e2e_tools_client.py
Os exemplos acima usam stdio. Troque o transporte para falar com servidores remotos (veja Transportes).
Conceitos Principais
Ferramentas
Descubra e chame funções expostas pelo 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)
Veja o exemplo completo: examples/e2e_tools_client.py
Recursos
Liste/leia (e opcionalmente assine) fontes de dados.
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)
Veja exemplos completos:
Prompts
Modelos de prompt parametrizados e reutilizáveis.
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)
Veja o exemplo completo: examples/e2e_prompts_client.py
Raízes (opcional)
Anuncie diretórios que o cliente autoriza o servidor a acessar.
from chuk_mcp.protocol.messages.roots import send_roots_list
roots = await send_roots_list(read, write) # if supported
Veja o exemplo completo: examples/e2e_roots_client.py
Amostragem e Conclusão (opcional)
Alguns servidores podem pedir ao cliente para amostrar texto ou fornecer conclusão para argumentos. Esses são opt-in e controlados por capacidade.
Veja exemplos completos:
Transportes
chuk-mcp separa claramente protocolo de transporte, para que você possa usar os mesmos manipuladores de protocolo com qualquer camada de transporte:
- Stdio — ideal para servidores de processos filhos locais
- HTTP Streamable — fale com servidores remotos via HTTP (chunked/NDJSON)
- SSE (Server-Sent Events) — para integrações com navegador/IDE com push unidirecional do servidor
- Extensível — implemente seu próprio transporte adaptando a interface assíncrona simples
(read, write)
Nota: O chuk-mcp é totalmente assíncrono (AnyIO). Use
anyio.run(...)ou integre ao seu loop de eventos.
Nota: As capacidades do protocolo são negociadas durante
initialize, independentemente do transporte. Você escolhe o transporte (stdio ou HTTP Streamable) com base nas necessidades de implantação/runtime.
Segurança de threads: Instâncias de cliente não são seguras para threads entre loops de eventos. Veja FAQ para detalhes.
HTTP Streamable usa NDJSON em chunks. Configure
HttpClientParameters(timeout_s=30, headers={"Authorization": "Bearer ..."}). Clientes transmitem NDJSON com backpressure. Para payloads grandes, prefira chunks NDJSON em vez de blobs base64 para evitar picos de memória.
Enquadramento: HTTP Streamable usa NDJSON (um objeto JSON por linha). Servidores devem fazer flush após cada objeto; proxies não devem armazenar em buffer indefinidamente.
Compressão: Ative gzip no proxy para reduzir grandes fluxos de conteúdo. Payloads MCP comprimem bem.
Design da Camada de Protocolo: A camada de protocolo é intencionalmente limpa e mínima — erros são levantados imediatamente sem tentativas. Esse design mantém a camada de protocolo focada no transporte de mensagens e na conformidade com a especificação MCP. Para casos de uso que exigem lógica de tentativa, tratamento de erros, limitação de taxa ou cache, use chuk-tool-processor, que fornece wrappers componíveis para tentativas com backoff exponencial, limitação de taxa e cache. Essa separação de preocupações permite que você escolha a estratégia de tentativa certa para as necessidades específicas da sua aplicação.
Segurança: Ao expor HTTP Streamable, termine o TLS em um proxy e exija autenticação (por exemplo, tokens bearer). Para CAs privadas, configure o trust store do seu cliente (por exemplo,
SSL_CERT_FILE=/path/ca.pem,REQUESTS_CA_BUNDLEouSSL_CERT_DIR). A camada de protocolo é agnóstica de transporte e não impõe autenticação.
Exemplos de Configuração
Config JSON (o cliente decide como iniciar/conectar)
{
"mcpServers": {
"sqlite": {
"command": "uvx",
"args": ["mcp-server-sqlite", "--db-path", "database.db"]
},
"filesystem": {
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "."]
}
}
}
Carregando config em 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)
Exemplos e Demonstrações de Recursos
O diretório examples/ contém demonstrações abrangentes e funcionais de todos os recursos do MCP:
Exemplos de Início Rápido
quickstart_minimal.py— Configuração mínima de cliente MCPquickstart_sqlite.py— Trabalhando com servidor MCP SQLitequickstart_resources.py— Acessando recursos do servidorquickstart_complete.py— Demonstração de múltiplos recursos
Exemplos Ponta a Ponta (E2E)
Pares completos cliente-servidor construídos com chuk-mcp puro, demonstrando tanto a implementação do cliente quanto do servidor para cada recurso do MCP:
Recursos Principais:
e2e_tools_client.py— Registro, descoberta e invocação de ferramentase2e_resources_client.py— Listagem e leitura de recursose2e_prompts_client.py— Modelos de prompt reutilizáveis
Recursos Avançados:
e2e_roots_client.py— Gerenciamento de raízes do sistema de arquivose2e_sampling_client.py— Requisições LLM iniciadas pelo servidore2e_completion_client.py— Funcionalidade de autocompletare2e_subscriptions_client.py— Notificações de mudança de recursose2e_cancellation_client.py— Cancelamento de operaçõese2e_progress_client.py— Acompanhamento de progressoe2e_logging_client.py— Tratamento de mensagens de loge2e_elicitation_client.py— Requisições de entrada do usuárioe2e_annotations_client.py— Metadados de conteúdo
Tratamento de Erros:
initialize_error_handling.py— Padrões abrangentes de tratamento de erros (OAuth 401, incompatibilidade de versão, timeout, etc.)
Executando Exemplos:
Muitos exemplos E2E são autocontidos com seu próprio servidor de nível de protocolo construído usando chuk-mcp puro. Quando relevante, o cliente inicia o servidor de demonstração correspondente:
# 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: Quando relevante, os exemplos incluem um
e2e_*_server.pycorrespondente mostrando um servidor mínimo construído com a mesma camada de protocolo.
Consulte examples/README.md para documentação detalhada de todos os exemplos.
Versionamento e Compatibilidade
chuk-mcpsegue as revisões da especificação MCP e negocia capacidades na inicialização.- Recursos mais novos são controlados por capacidade e degradam graciosamente com servidores mais antigos.
- Tipagem/validação opcional usa Pydantic se disponível, caso contrário, um fallback leve.
📋 Versões de Protocolo Suportadas (a partir de v0.1.x)
| Versão | Status | Política de Suporte |
|---|---|---|
2025-06-18 | Mais recente | Suporte primário, todos os recursos |
2025-03-26 | Estável | Compatibilidade total, mantida |
2024-11-05 | Legado | Compatibilidade reversa, depreciação a definir |
Plataformas Testadas: Linux, macOS, Windows (Python 3.11+)
Política de Suporte: As versões Mais Recente e Estável recebem suporte total. O suporte à versão Legado será mantido até 2026-Q2, após o qual poderá ser depreciado. Consulte o changelog para orientações de migração.
📊 Matriz de Suporte de Recursos do Cliente
| Categoria de Recurso | 2024-11-05 | 2025-03-26 | 2025-06-18 | Status de Implementação |
|---|---|---|---|---|
| Operações Principais | ||||
| Ferramentas (listar/chamar) | ✅ | ✅ | ✅ | ✅ Completo |
| Recursos (listar/ler/assinar) | ✅ | ✅ | ✅ | ✅ Completo |
| Prompts (listar/obter) | ✅ | ✅ | ✅ | ✅ Completo |
| Transporte | ||||
| Stdio | ✅ | ✅ | ✅ | ✅ Completo |
| HTTP Transmissível | – | ✅ | ✅ | ✅ Completo |
| Recursos Avançados | ||||
| Amostragem | ✅ | ✅ | ✅ | ✅ Completo |
| Conclusão | ✅ | ✅ | ✅ | ✅ Completo |
| Raízes | ✅ | ✅ | ✅ | ✅ Completo |
| Elicitação | ❌ | ❌ | ✅ | ✅ Completo |
| Recursos de Qualidade | ||||
| Rastreamento de Progresso | ✅ | ✅ | ✅ | ✅ Completo |
| Cancelamento | ✅ | ✅ | ✅ | ✅ Completo |
| Notificações | ✅ | ✅ | ✅ | ✅ Completo |
| Registro | ✅ | ✅ | ✅ | ✅ Completo |
| Anotações | ✅ | ✅ | ✅ | ✅ Completo |
Os recursos degradam graciosamente ao interagir com servidores mais antigos.
Consulte o changelog para as versões exatas da especificação suportadas e quaisquer depreciações.
Política de Versionamento
Este projeto segue Versionamento Semântico para APIs públicas sob chuk_mcp.*:
- Maior (X.0.0): Mudanças que quebram compatibilidade nas APIs públicas
- Menor (0.X.0): Novos recursos, compatível com versões anteriores
- Correção (0.0.X): Correções de bugs, compatível com versões anteriores
Mudanças que Quebram Compatibilidade e Migração
v0.7.2: Mudanças no Tratamento de Exceções
O que Mudou: send_initialize() e send_initialize_with_client_tracking() agora sempre lançam exceções em vez de retornar None em erros.
Porquê: Isso permite tratamento adequado de erros, reautenticação automática OAuth em ferramentas downstream (como mcp-cli), e segue as melhores práticas do Python.
Guia de Migração:
Antes (v0.7.1 e 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}")
Depois (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}")
Mudanças no Tipo de Retorno:
send_initialize():Optional[InitializeResult]→InitializeResultsend_initialize_with_client_tracking():Optional[InitializeResult]→InitializeResult
Benefícios:
- ✅ Reautenticação automática OAuth no mcp-cli
- ✅ Propagação adequada de erros e depuração
- ✅ Segurança de tipos (sem necessidade de verificações de
Optional) - ✅ Contexto completo de exceção com rastreamentos de pilha
Veja Também:
EXCEPTION_HANDLING_FIX.md- Documentação técnica detalhadaexamples/initialize_error_handling.py- Exemplos completos de tratamento de errostests/mcp/messages/test_initialize_exceptions.py- Suíte de testes
Comparação com o SDK Oficial MCP
| Recurso | chuk-mcp | SDK Python Oficial MCP |
|---|---|---|
| Filosofia | Biblioteca de conformidade de protocolo | Framework completo |
| Escopo | Cliente + Servidor, focado em protocolo | Framework Cliente + Servidor |
| Tipagem | Pydantic opcional (fallback disponível) | Pydantic obrigatório |
| Transportes | stdio, SSE, HTTP Transmissível (plugável) | stdio, SSE, HTTP Transmissível |
| Navegador/WASM | Compatível com Pyodide | Varia / não é alvo principal |
| Dependências | Mínimas (núcleo anyio) | Pilha mais pesada |
| Framework de Servidor | Auxiliares leves | Estrutura de servidor opinativa |
| Estilo de API | Funções explícitas send_* | Abstrações de nível superior |
| Caso de Uso Alvo | Integração de protocolo, clientes/servidores personalizados | Aplicações MCP completas |
| Orquestração | Externa (você escolhe) | Padrões integrados |
| Curva de Aprendizado | Baixa (nível de protocolo) | Média (conceitos de framework) |
Quando escolher chuk-mcp:
- Construindo clientes ou servidores MCP personalizados
- Precisa de flexibilidade de transporte (HTTP Transmissível)
- Quer dependências mínimas
- Prefere controle no nível de protocolo
- Executando em ambientes restritos (WASM, funções de borda)
- Precisa integrar MCP em aplicações existentes
Exemplo do mundo real: chuk-mcp-server usa chuk-mcp como sua camada de conformidade de protocolo
Quando escolher o SDK oficial:
- Construindo servidores MCP completos rapidamente com padrões opinativos
- Quer abstrações de framework prontas para uso
- Usando principalmente transporte stdio
- Prefere APIs de nível superior
Objetivos de Design e Não-Objetivos
Objetivos
- Ser a maneira mais simples de implementar MCP em Python (cliente ou servidor)
- Manter a API pequena, explícita e tipada
- Tornar transportes plugáveis e a lógica de protocolo reutilizável
- Suportar casos de uso de cliente e servidor com abstrações leves
Não-Objetivos
- Competir com frameworks completos de agentes / IDEs
- Incorporar estrutura de aplicação opinativa ou mecanismos de fluxo de trabalho
- Enviar dependências pesadas por padrão
- Fornecer orquestração de alto nível (essa é sua camada de aplicação)
Escalabilidade e Concorrência
chuk-mcp lida com centenas de conexões concorrentes eficientemente com uso mínimo de recursos:
Benchmarks de Concorrência
Desempenho Testado (veja benchmarks/PERFORMANCE_REPORT.md para detalhes completos):
- 700+ conexões concorrentes testadas com sucesso (parou por timeout, não por limite de capacidade)
- 252+ conexões/segundo de throughput para rotatividade rápida de conexões
- ~34KB de memória por conexão com escalabilidade linear
- Zero vazamentos de memória verificados ao longo de 200+ iterações
Estimativas de Capacidade:
- Pequena escala (< 100 agentes): 512MB de RAM, 1 núcleo
- Média escala (100-1.000 agentes): 1-2GB de RAM, 2-4 núcleos
- Grande escala (1.000-10.000 agentes): 4-8GB de RAM, 8+ núcleos
- Escala empresarial (10.000+ agentes): Balanceamento de carga recomendado
Melhores Práticas
Padrão: Criar todos → Inicializar todos (Sequencial)
# 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()
Padrão: Intercalado (Também Suportado)
# 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: Sempre use StdioClient como um gerenciador de contexto assí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
Recomendações de Monitoramento
Para implantações, monitore estas métricas:
- Conexões Ativas: Acompanhe a contagem de clientes concorrentes
- Crescimento de Memória: Deve permanecer estável ao longo do tempo (~0,034MB por conexão)
- Descritores de Arquivo: Monitore via
lsofou/proc/<pid>/fd - Taxa de Sucesso de Conexão: Deve manter 100%
Consulte benchmarks/PERFORMANCE_REPORT.md para análise detalhada de desempenho e diretrizes de implantação.
FAQ
P: Isso inclui um framework de servidor?
R: Sim, na camada de protocolo. chuk-mcp fornece mensagens tipadas e auxiliares utilizáveis tanto em clientes quanto em servidores, mas não é um framework de servidor opinativo—você traz sua própria estrutura de aplicação/orquestração.
P: Pydantic é obrigatório?
R: Não. Se instalado (apenas Pydantic v2), você obterá tipos e validação mais ricos. Se não, a biblioteca usa um fallback leve com modelos baseados em dict.
P: Qual transporte devo usar?
R: Use stdio para desenvolvimento local e processos filhos. Use HTTP Transmissível para servidores remotos atrás de TLS com autenticação.
P: Onde posso encontrar mais exemplos?
R: Consulte o diretório examples/ para demonstrações abrangentes de todos os recursos MCP, incluindo exemplos de início rápido e pares completos cliente-servidor de ponta a ponta. Para uma implementação de servidor do mundo real, veja chuk-mcp-server que usa chuk-mcp como sua biblioteca de protocolo.
P: Como testo minha implementação?
R: Execute make test ou uv run pytest para rodar a suíte de testes. Use make examples (se presente) para testar todos os exemplos E2E. Consulte a seção Contribuindo para detalhes.
P: Isso está pronto para uso?
R: Sim. chuk-mcp está implantado em escala. Inclui tratamento de erros, segurança de tipos e segue as especificações do protocolo MCP. Consulte os relatórios de cobertura de testes para métricas de confiança.
P: É seguro para threads?
R: Instâncias de cliente não são seguras para threads entre loops de eventos. Compartilhe um cliente dentro de um único loop assíncrono; use instâncias separadas por loop/thread.
P: O que não está incluído?
R: Autenticação, terminação TLS, persistência e orquestração são preocupações da aplicação—traga as suas próprias. chuk-mcp fornece apenas conformidade de protocolo. Para frontends de navegador/WASM com CORS e TLS, termine o TLS no proxy e defina Access-Control-Allow-Origin para a origem do seu frontend; evite * com credenciais.
P: Como adiciono lógica de repetição e limitação de taxa?
R: Use chuk-tool-processor que fornece wrappers componíveis para repetições (com backoff exponencial), limitação de taxa e cache. chuk-mcp foca na conformidade de protocolo; chuk-tool-processor lida com preocupações de execução.
P: Quais são os erros comuns e como lidar com eles?
R: Exceções comuns e ações recomendadas:
| Tipo de Erro | Código JSON-RPC | Ação |
|---|---|---|
| Erro de análise | -32700 | Corrija a sintaxe JSON na solicitação |
| Solicitação inválida | -32600 | Verifique os campos obrigatórios (jsonrpc, method, id) |
| Método não encontrado | -32601 | Verifique o nome do método e as capacidades do servidor |
| Parâmetros inválidos | -32602 | Valide os tipos de parâmetros e campos obrigatórios |
| Erro interno | -32603 | Verifique os logs do servidor, repita a operação |
| Erro de autenticação (401) | -32603 | Reautentique (automático no mcp-cli) |
| Solicitação cancelada | -32800 | Lide com o cancelamento graciosamente |
| Conteúdo muito grande | -32801 | Reduza o tamanho do payload ou use streaming |
| Conexão/Transporte | varia | Verifique a rede, confirme que o servidor está em execução |
Nota: Ao usar transportes baseados em HTTP (SSE ou HTTP Transmissível), erros na camada de transporte (falhas de rede, problemas TLS, problemas de autenticação) aparecerão como códigos de status HTTP antes de atingir a camada de protocolo MCP. No entanto, uma vez que o transporte é estabelecido, todos os erros de protocolo MCP seguem o sistema de códigos de erro JSON-RPC mostrado acima.
Todos os erros de protocolo herdam de classes de exceção base e são sempre lançados (nunca retornam None). Consulte exemplos para padrões de tratamento de erros.
Melhores Práticas de Tratamento de Exceções:
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 demonstrações abrangentes de tratamento de erros.
Contribuindo
PRs são bem-vindos! Por favor:
- Abra um issue pequeno e focado primeiro (opcional, mas útil).
- Adicione testes e dicas de tipo para novas funcionalidades.
- Mantenha APIs públicas mínimas e consistentes.
- Execute os linters e a suíte de testes antes de enviar.
PRs devem manter cobertura ≥85%; aplicado no CI junto com verificações de tipo mypy e 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
Relatórios de bugs / solicitações de recursos: Modelos de issue disponíveis em
.github/
Código de Conduta: Espera-se que os contribuidores sigam o Pacto do Contribuidor
Segurança
Se você acredita ter encontrado um problema de segurança, por favor reporte abrindo um advisory de segurança no repositório GitHub em vez de abrir um issue público.
Vitrine de Recursos
Esta seção fornece snippets de código detalhados demonstrando os recursos do MCP. Todos os exemplos incluem segurança total de tipos.
🔧 Ferramentas — Chamando Funções
Ferramentas são funções que a IA pode 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}")
Exemplo completo: uv run python examples/e2e_tools_client.py
📄 Recursos — Lendo Dados
Recursos fornecem acesso a fontes de dados (arquivos, bancos de dados, 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]}...")
Exemplo completo: uv run python examples/e2e_resources_client.py
📡 Assinaturas de Recursos — Atualizações em Tempo Real
Assine recursos para receber notificações de mudanças em tempo 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")
Exemplo completo: uv run python examples/e2e_subscriptions_client.py
💬 Prompts — Gerenciamento de Modelos
Prompts são modelos reutilizáveis com 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}")
Exemplo completo: uv run python examples/e2e_prompts_client.py
🎯 Amostragem — Geração de Conteúdo por IA
Permite que servidores solicitem à IA que gere conteúdo em seu nome (requer aprovação do usuário):
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: Servidores podem usar amostragem para gerar código, documentação ou análise com base nos dados aos quais têm acesso.
Exemplo completo: uv run python examples/e2e_sampling_client.py
📁 Raízes — Controle de Acesso a Diretórios
Raízes definem quais diretórios o cliente permite que os servidores acessem.
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: Controle quais diretórios a IA pode acessar, permitindo operações seguras em sandbox.
Exemplo completo: uv run python examples/e2e_roots_client.py
🎭 Elicitação — Solicitações de Entrada do Usuário
A elicitação permite que servidores solicitem entrada estruturada dos usuários:
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: Fluxos de trabalho interativos, fluxos OAuth, diálogos de confirmação.
Exemplo completo: uv run python examples/e2e_elicitation_client.py
💡 Conclusão — Autocompletar Inteligente
Obtenha sugestões inteligentes para argumentos de ferramentas:
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}")
Exemplo completo: uv run python examples/e2e_completion_client.py
📊 Acompanhamento de Progresso
Monitore operações de longa duração com atualizações de progresso:
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
Exemplo completo: uv run python examples/e2e_progress_client.py
🚫 Cancelamento
Cancele operações de longa duração com timeout:
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)
Exemplo completo: uv run python examples/e2e_cancellation_client.py
🌐 Múltiplos Transportes
Use diferentes protocolos de transporte para diferentes cenários:
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)
🔄 Orquestração Multi-Servidor
Conecte-se a vários servidores simultaneamente:
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}")
Segurança de Tipos e Validação
Todas as mensagens do protocolo retornam resultados totalmente tipados usando Pydantic (ou validação de fallback):
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)
Benefícios:
- Retornos tipados: Todas as funções
send_*retornam modelos Pydantic tipados - Análise de conteúdo: Use
parse_content()para manipulação de conteúdo com segurança de tipos - Validação em tempo de execução: Validação automática com mensagens de erro claras
- Suporte a IDE: Autocompletar completo e verificação de tipos
Monitoramento e Registro
Recursos integrados para ambientes de produção:
from chuk_mcp.protocol.messages.logging import send_logging_set_level
# Set server logging level
await send_logging_set_level(write, level="debug")
Recursos:
- Registro estruturado com níveis configuráveis
- Monitoramento de desempenho (latência, taxas de erro, throughput)
- Suporte a acompanhamento de progresso e cancelamento
- Propagação de erros limpa (sem novas tentativas automáticas na camada de protocolo)
Exemplo completo: uv run python examples/e2e_logging_client.py
Ecossistema
chuk-mcp faz parte de um conjunto modular de ferramentas MCP em Python:
- chuk-tool-processor — Execução confiável de chamadas de ferramentas com novas tentativas, cache e backoff exponencial
- chuk-mcp-server — Implementação real de servidor MCP construída sobre chuk-mcp
- chuk-mcp-cli — CLI interativo e playground para testar servidores MCP
Cada componente foca em fazer uma coisa bem e pode ser usado de forma independente ou em conjunto. Todos eles se baseiam na camada de protocolo do chuk-mcp, portanto herdam as mesmas características de baixa latência e sobrecarga mínima.
Licença
Apache 2.0 — veja LICENSE.