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

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

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:

CamadaPropósitoUso
CLI e DemonstraçãoUtilitários integrados e demonstraçõesOpcional — use a camada de protocolo diretamente
API de Cliente/ServidorAbstrações de alto nível para interações cliente-servidorOpcional — pode usar a camada de protocolo diretamente
Camada de ProtocoloDefinições de mensagens, tratamento de requisições/respostas com segurança de tipos, negociação de capacidadesNúcleo — implementa a especificação MCP
Camada de TransporteImplementações de transporte plugáveis (stdio, HTTP Streamable)Escolha com base na implantação
Camada BaseFallback Pydantic, configuração compartilhada, adaptadores de tipoFundaçã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?

  • 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,<1 e orjson>=3.10.0,<4 para 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 run ou 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_BUNDLE ou SSL_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

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:

Recursos Avançados:

Tratamento de Erros:

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.py correspondente 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-mcp segue 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ãoStatusPolítica de Suporte
2025-06-18Mais recenteSuporte primário, todos os recursos
2025-03-26EstávelCompatibilidade total, mantida
2024-11-05LegadoCompatibilidade 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 Recurso2024-11-052025-03-262025-06-18Status 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] → InitializeResult
  • send_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:


Comparação com o SDK Oficial MCP

Recursochuk-mcpSDK Python Oficial MCP
FilosofiaBiblioteca de conformidade de protocoloFramework completo
EscopoCliente + Servidor, focado em protocoloFramework Cliente + Servidor
TipagemPydantic opcional (fallback disponível)Pydantic obrigatório
Transportesstdio, SSE, HTTP Transmissível (plugável)stdio, SSE, HTTP Transmissível
Navegador/WASMCompatível com PyodideVaria / não é alvo principal
DependênciasMínimas (núcleo anyio)Pilha mais pesada
Framework de ServidorAuxiliares levesEstrutura de servidor opinativa
Estilo de APIFunções explícitas send_*Abstrações de nível superior
Caso de Uso AlvoIntegração de protocolo, clientes/servidores personalizadosAplicações MCP completas
OrquestraçãoExterna (você escolhe)Padrões integrados
Curva de AprendizadoBaixa (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 lsof ou /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 ErroCódigo JSON-RPCAção
Erro de análise-32700Corrija a sintaxe JSON na solicitação
Solicitação inválida-32600Verifique os campos obrigatórios (jsonrpc, method, id)
Método não encontrado-32601Verifique o nome do método e as capacidades do servidor
Parâmetros inválidos-32602Valide os tipos de parâmetros e campos obrigatórios
Erro interno-32603Verifique os logs do servidor, repita a operação
Erro de autenticação (401)-32603Reautentique (automático no mcp-cli)
Solicitação cancelada-32800Lide com o cancelamento graciosamente
Conteúdo muito grande-32801Reduza o tamanho do payload ou use streaming
Conexão/TransportevariaVerifique 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:

  1. Abra um issue pequeno e focado primeiro (opcional, mas útil).
  2. Adicione testes e dicas de tipo para novas funcionalidades.
  3. Mantenha APIs públicas mínimas e consistentes.
  4. 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.