chuk-mcp
About
A Python client for the Model Context Protocol (MCP), an open standard for connecting AI assistants to external data and tools.
Explore
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)
# Install SQLite server uv tool install mcp-server-sqlite # Run example uv run python examples/quickstart_sqlite.py
Build your own MCP server using the same protocol layer. Seeexamples/e2e__server.pyfor complete working servers:
# 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)
# See examples/ for complete client-server pairs uv run python examples/e2e_tools_client.py
The examples above use stdio. Swap the transport to talk to remote servers (seeTransports).
Discover and call server-exposed functions.
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)
See full example:examples/e2e_tools_client.py
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)
- examples/e2e_resources_client.py
- examples/e2e_subscriptions_client.py
Parameterized, reusable prompt templates.
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)
See full example:examples/e2e_prompts_client.py
Advertise directories the client authorizes the server to access.
from chuk_mcp.protocol.messages.roots import send_roots_list roots = await send_roots_list(read, write) # if supported
See full example:examples/e2e_roots_client.py
Some servers can ask the client to sample text or provide completion for arguments. These are opt-in and capability-gated.
- examples/e2e_sampling_client.py
- examples/e2e_completion_client.py
chuk-mcpcleanly separatesprotocolfromtransport, so you can use the same protocol handlers with any transport layer:
- Stdio— ideal for local child-process servers
- Streamable HTTP— speak to remote servers over HTTP (chunked/NDJSON)
- SSE (Server-Sent Events)— for browser/IDE integrations with one-way server push
- Extensible— implement your own transport by adapting the simple(read, write)async interface
Note:chuk-mcp is fully async (AnyIO). Useanyio.run(...)or integrate into your event loop.
Note:Protocolcapabilitiesare negotiated duringinitialize, independent of transport. You choose the transport (stdio or Streamable HTTP) based on deployment/runtime needs.
Thread-safety:Client instances are not thread-safe across event loops. SeeFAQfor details.
Streamable HTTPuses chunked NDJSON. ConfigureHttpClientParameters(timeout_s=30, headers={"Authorization": "Bearer ..."}). Clients stream NDJSON with backpressure. For large payloads, prefer NDJSON chunks over base64 blobs to avoid memory spikes.
Framing:Streamable HTTP uses NDJSON (one JSON object per line). Servers should flush after each object; proxies must not buffer indefinitely.
Compression:Enable gzip at the proxy to reduce large content streams. MCP payloads compress well.
Protocol Layer Design:The protocol layer is intentionallyclean and minimal— errors are raised immediately without retries. This design keeps the protocol layer focused on message transport and compliance with the MCP specification. For use cases requiring retry logic, error handling, rate limiting, or caching, use[chuk-tool-processorwhich provides composable wrappers for retries with exponential backoff, rate limiting, and caching. This separation of concerns allows you to choose the right retry strategy for your specific application needs.
Security:When exposing Streamable HTTP, terminate TLS at a proxy and require auth (e.g., bearer tokens). For private CAs, configure your client's trust store (e.g.,SSL_CERT_FILE=/path/ca.pem,REQUESTS_CA_BUNDLE, orSSL_CERT_DIR). The protocol layer is transport-agnostic and does not impose auth.
{ "mcpServers": { "sqlite": { "command": "uvx", "args": ](https://pypi.org/project/chuk-tool-processor/)["mcp-server-sqlite", "--db-path", "database.db"] }, "filesystem": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "."] } } }
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)
Theexamples/directory contains comprehensive, working demonstrations of all MCP features:
- quickstart_minimal.py— Minimal MCP client setup
- quickstart_sqlite.py— Working with SQLite MCP server
- quickstart_resources.py— Accessing server resources
- quickstart_complete.py— Multi-feature demo
Completeclient-server pairsbuilt with pure chuk-mcp, demonstrating both client and server implementation for each MCP feature:
- e2e_tools_client.py— Tool registration, discovery, and invocation
- e2e_resources_client.py— Resource listing and reading
- e2e_prompts_client.py— Reusable prompt templates
- e2e_roots_client.py— File system root management
- e2e_sampling_client.py— Server-initiated LLM requests
- e2e_completion_client.py— Autocomplete functionality
- e2e_subscriptions_client.py— Resource change notifications
- e2e_cancellation_client.py— Operation cancellation
- e2e_progress_client.py— Progress tracking
- e2e_logging_client.py— Log message handling
- e2e_elicitation_client.py— User input requests
- e2e_annotations_client.py— Content metadata
A lean, minimal Python implementation of the Model Context Protocol (MCP).
Brings first-class MCP protocol support to Python — lightweight, async, and spec-accurate from day one.
chuk-mcpgives you a clean, typed, transport-agnostic implementation for bothMCP clients and servers. It focuses on the protocol surface (messages, types, versioning, transports) and leaves orchestration, UIs, and agent frameworks to other layers.
✳️What this is: aprotocol compliance librarywith ergonomic helpers for clients and servers.
⛔What this isn't: a chatbot runtime, workflow engine, or an opinionated application framework.
┌──────────────────────────────────────┐ │ Your AI Application │ │ (Claude, GPT, custom agents) │ └────────────┬─────────────────────────┘ │ MCP Protocol ▼ ┌──────────────────────────────────────┐ │ chuk-mcp Client │ ← You are here │ • Protocol compliance │ │ • Transport (stdio/Streamable HTTP)│ │ • Type-safe messages │ │ • Capability negotiation │ └────────────┬─────────────────────────┘ │ MCP Protocol ▼ ┌──────────────────────────────────────┐ │ chuk-mcp Server (optional) │ │ • Protocol handlers │ │ • Tool/Resource registration │ │ • Session management │ └────────────┬─────────────────────────┘ │ ▼ ┌──────────────────────────────────────┐ │ Your Tools & Resources │ │ (databases, APIs, files, etc) │ └──────────────────────────────────────┘
chuk-mcp provides the protocol layer— connect AI applications to tools and data sources using the standard MCP protocol.
The library itself is organized in layers that you can use at different levels of abstraction:
┌─────────────────────────────────────────┐ │ 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 └─────────────────────────────────────────┘
Most users work with theProtocol Layer(send_functions) andTransport Layer(stdio/HTTP clients), optionally using theClient/Server APIfor higher-level abstractions.
- Why chuk‑mcp?
- Protocol Performance
- At a Glance
- Install
- Quick Start
- Core Concepts
- Tools
- Resources
- Prompts
- Roots (optional)
- Sampling & Completion (optional)
- Protocol-first: Focuses on MCP messages, types, and capability negotiation —spec.modelcontextprotocol.io
- Client + Server: Full support for building both MCP clients and servers
- Typed: Full type hints; optional Pydantic models when available
- Transport-agnostic: stdio by default, Streamable HTTP (NDJSON) for remote servers, easily extensible
- Async-first: Built on AnyIO; integrate withanyio.run(...)or your existing loop
- Small & focused: No heavy orchestration or agent assumptions
- Clean protocol layer: Errors fail fast without retries — bring your own error handling strategy
- Reliable: Clear errors, structured logging hooks, composable with retry/caching layers
- ⚡ High-performance: Protocol overhead in the 2-5ms range; optional fast JSON for 4x faster serialization. SeeProtocol Performancefor detailed benchmarks
chuk-mcpis designed to keep MCP protocol overhead in the2-5 msrange, so the cost of using tools is dominated by the tools themselves, not the protocol.
- Zero heavy dependencies (AnyIO core only)
- Async-native stdio & NDJSON HTTP
- No tool execution inside the library
- Optional orjson fast path ([fast-json])
💡 For concurrency & capacity numbers, seeScaling & Concurrency.
Protocol overhead (typical measurements on modern hardware):
- Initialize → Tool List:2-3 ms
- Tool Call Round Trip:< 5 ms overhead (beyond actual tool execution time)
- Streaming:Near-zero overhead due to NDJSON chunk boundaries
Benchmarks run on macOS (Darwin 24.6.0), Python 3.11 — seebenchmarks/PERFORMANCE_REPORT.mdfor exact environment and commands.
🚀 JSON Serialization (Optional Fast Path)
Install with[fast-json]for~4x faster JSON operationsusing orjson:
- Serialization:~6x faster
- Deserialization:~2x faster
- Round-trip:~4x faster
pip install "chuk-mcp[fast-json]" # Automatic with graceful fallback
Benchmark numbers frombenchmarks/json_performance.pycomparing orjson vs stdlib json on realistic MCP messages.*
- High-frequency tool calls— minimal overhead per request
- Real-time agents— sub-5ms protocol latency
- Streaming UIs— near-zero NDJSON chunk overhead
- Tool processors— fast enough to be transparent
- WASM/edge environments— minimal footprint
- High-throughput workloads— proven at scale (seeScaling & Concurrency)
# Install an example MCP server uv tool install mcp-server-sqlite # Run the quick-start example uv run python examples/quickstart_sqlite.py
A minimal working MCP server in ~10 lines:
# 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)
Run it:uv run python hello_mcp.py— or connect any MCP client via stdio!
# 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)
# 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)
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
…
Sign in to leave a review
Use Google, GitHub, or an email account so ratings stay tied to real people.
No reviews posted yet.
