┌──────────────────────────────────────┐
│   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)      │
└──────────────────────────────────────┘

┌─────────────────────────────────────────┐
│              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
└─────────────────────────────────────────┘

pip install "chuk-mcp[fast-json]"  # Automatic with graceful fallback

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

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

# 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)

# 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)

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)

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)

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)

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

# 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

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

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}")

# 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()

# 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()

# 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

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}")

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]}...")

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")

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}")

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'}")

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")

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}")

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}")

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

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)

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)

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}")

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)

from chuk_mcp.protocol.messages.logging import send_logging_set_level

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

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

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

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

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)

uv run python examples/quickstart_minimal.py

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

# 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

{
  "mcpServers": {
    "sqlite": {
      "command": "uvx",
      "args": ["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)

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}")