1. Definitions:
- MCP (Model Context Protocol): An open, JSON-RPC 2.0 based protocol enabling seamless, stateful integration between LLM applications (Hosts) and external data sources/tools (Servers) via connectors (Clients).
- Host: The main LLM application (e.g., IDE, chat interface) that manages Clients and user interaction.
- Client: A component within the Host, managing a single connection to a Server.
- Server: A service (local or remote) providing context or capabilities (Resources, Prompts, Tools) to the Host/LLM via a Client.
2. Philosophy & Design Principles:
- Simplicity: Servers should be easy to build, focusing on specific capabilities.
- Composability: Multiple focused Servers can be combined for complex workflows.
- Isolation: Servers operate independently, without access to the full conversation or other servers' data. Host enforces boundaries.
- Progressive Features: Core functionality is minimal; advanced features (capabilities) are negotiated.
- User Control & Safety: Emphasizes explicit user consent, data privacy, tool safety (human-in-the-loop), and controlled LLM sampling.
3. Key Features & Capabilities:
- Base: JSON-RPC 2.0 messages (Requests, Responses, Notifications), stateful connections, capability negotiation.
- Lifecycle: initialize (negotiate version/capabilities) -> initialized -> Operation -> Shutdown.
- Server Features:
- Resources: Expose contextual data (files, DB schemas) via URIs (resources/list, resources/read, resources/subscribe).
- Prompts: Offer user-selectable templates/workflows (prompts/list, prompts/get).
- Tools: Expose LLM-callable functions with JSON schema (tools/list, tools/call). Annotations provide hints (e.g., readOnlyHint).
- Client Features:
- Sampling: Allow Servers to request LLM generation via the Client (sampling/createMessage).
- Roots: Expose filesystem boundaries (roots/list).
- Utilities: ping, notifications/cancelled, notifications/progress, logging/setLevel, notifications/message, completion/complete, Pagination (cursor-based).
- Transport: stdio (recommended), Streamable HTTP (POST/GET, optional SSE, sessions).
- Authentication (HTTP): OAuth 2.1 framework, Metadata Discovery, Dynamic Client Registration.
4. Best Practices:
- Security: Implement robust user consent flows. Validate URIs/inputs/outputs. Sanitize data. Use HTTPS. Follow OAuth 2.1 security guidelines. Treat tool annotations from untrusted servers with caution.
- Human-in-the-Loop: Always require user confirmation for tool execution and sampling requests. Allow prompt review/editing.
- Implementation: Support base protocol and lifecycle. Implement needed capabilities. Handle errors and timeouts gracefully. Use latest protocol version. Treat pagination cursors as opaque.
- Transport: Prefer stdio for local subprocesses. Use specified Auth for HTTP.
5. Example Minimum Implementation:
- Transport: Implement stdio (read stdin lines, write stdout lines).
- Lifecycle:
- Handle initialize request: Check protocolVersion, respond with server capabilities (e.g., {}) and serverInfo.
- Handle notifications/initialized notification: Transition to operational state.
- Core: Implement ping request handler (respond with empty result).
- JSON-RPC: Parse incoming JSON, validate structure, dispatch to handlers. Format outgoing JSON messages correctly. Handle unique ids for requests/responses.
(One-Page Reference)
Overview:
Model Context Protocol (MCP) connects LLM applications (Hosts) with external tools/data (Servers) using JSON-RPC 2.0. This guide focuses on implementing MCP Servers and Clients in Python. The latest protocol version is 2025-03-26.
Core Concepts:
- Host: The main application (e.g., your Python app using an LLM). Manages Clients.
- Client: Connects the Host to one Server. Handles communication and capability negotiation. Often part of the Host.
- Server: Provides Resources, Prompts, or Tools. Can be a separate Python process/service.
Protocol Basics:
- Format: JSON-RPC 2.0 over UTF-8. Use standard libraries (json).
- Messages:
- Request: {"jsonrpc": "2.0", "id": 123, "method": "...", "params": {...}} (Requires Response)
- Response: {"jsonrpc": "2.0", "id": 123, "result": {...}} or {"jsonrpc": "2.0", "id": 123, "error": {"code": ..., "message": ...}}
- Notification: {"jsonrpc": "2.0", "method": "...", "params": {...}} (No Response, No ID)
- Batch: Arrays [...] of Requests/Notifications (sending) or Responses/Errors (receiving). Servers MUST support receiving batches.
- Versioning: YYYY-MM-DD strings. Negotiated during initialize. Prefer latest supported version.
Lifecycle:
- Initialization:
- Client sends initialize request (protocolVersion, clientInfo, capabilities).
- Server checks protocolVersion, responds with chosen protocolVersion, serverInfo, capabilities (or error).
- Client sends notifications/initialized notification.
- Constraint: initialize MUST NOT be batched.
- Operation: Exchange messages based on negotiated capabilities.
- Shutdown: Close transport connection (e.g., close streams for stdio, close HTTP connections/session). Implement timeouts for requests.
Key Server Features (Implement based on Server Capabilities):
- Resources (resources capability): Provide data context.
- resources/list: Return ListResourcesResult (paginated list of Resource objects).
- resources/read: Return ReadResourceResult (content as TextResourceContents or BlobResourceContents).
- resources/templates/list: List ResourceTemplates (URI templates).
- Optional: resources/subscribe (client asks for updates), notifications/resources/updated (server sends updates), notifications/resources/list_changed.
- Resource: { "uri": "...", "name": "...", "description": "...", "mimeType": "...", "annotations": {...} }
- Prompts (prompts capability): Offer pre-defined interactions.
- prompts/list: Return ListPromptsResult (paginated list of Prompt objects).
- prompts/get: Return GetPromptResult (list of PromptMessages based on name/args).
- Optional: notifications/prompts/list_changed.
- Prompt: { "name": "...", "description": "...", "arguments": [...] }
- PromptMessage: { "role": "user"|"assistant", "content": Text|Image|Audio|EmbeddedResource }
- Tools (tools capability): Allow LLM to execute actions.
- tools/list: Return ListToolsResult (paginated list of Tool objects).
- tools/call: Execute tool by name with arguments, return CallToolResult (content list, isError flag). Report tool execution errors within the result, not as protocol errors.
- Optional: notifications/tools/list_changed.
- Tool: { "name": "...", "description": "...", "inputSchema": {...}, "annotations": ToolAnnotations } (Use JSON Schema for inputSchema). Treat annotations as untrusted hints.
Key Client Features (Implement based on Client Capabilities):
- Sampling (sampling capability): Handle LLM generation requests from Servers.
- Handle sampling/createMessage requests. Prompt user for approval. Interact with LLM. Return CreateMessageResult.
- Use modelPreferences (priorities, hints) as advisory.
- Roots (roots capability): Provide filesystem context boundaries.
- Handle roots/list requests. Return ListRootsResult with Root objects ({"uri": "file://...", "name": "..."}).
- Optional: Send notifications/roots/list_changed when roots change.
Transports:
- stdio (Recommended for local):
- Server runs as subprocess.
- Read JSON messages line-by-line from sys.stdin.
- Write JSON messages line-by-line to sys.stdout. Add \n delimiter. Ensure UTF-8.
- Use sys.stderr for out-of-band logging.
- Must Not contain embedded newlines within a single JSON message string.
- Streamable HTTP:
- Server provides a single MCP endpoint path (e.g., /mcp).
- Client sends POST for requests/notifications/responses (single or batch). Expects application/json or text/event-stream (SSE).
- Client sends GET to listen for server-initiated messages via SSE (optional).
- Server MAY use Mcp-Session-Id header for stateful sessions (initiated in initialize response). Client MUST include it in subsequent requests.
- Optional: SSE id for resumability via Last-Event-ID.
Authentication (HTTP Only):
- Based on OAuth 2.1. Servers supporting Auth SHOULD implement Metadata Discovery (/.well-known/oauth-authorization-server) and SHOULD support Dynamic Client Registration (/register).
- Clients MUST support Metadata Discovery and use Bearer tokens (Authorization: Bearer ...) on every request.
- Fallbacks exist for non-discovery servers (/authorize, /token, /register).
- For stdio, use environment variables or other out-of-band methods for credentials.
Utilities (Optional):
- ping: Respond to ping requests with {}.
- notifications/cancelled: Handle cancellation notifications for in-progress requests. Send if client cancels.
- notifications/progress: Send progress updates (progressToken, progress, total, message) if requested via _meta.progressToken.
- logging/setLevel, notifications/message: Support structured logging (levels: debug, info, etc.).
- completion/complete: Provide argument autocompletion suggestions.
- Pagination: Support cursor in request params and nextCursor in results for list methods. Treat cursors as opaque.
Security:
- CRITICAL: Implement User Consent, Data Privacy, Tool Safety (Human-in-the-loop for tools/sampling), and Sampling Controls within the Host application. MCP itself does not enforce these.
- Validate all inputs (URIs, arguments, schemas). Sanitize outputs.
- Prevent path traversal (roots). Use HTTPS. Follow OAuth 2.1 best practices.
- Rate limit requests (completions, tools, logs). Avoid logging sensitive data.
Minimal Python Server Example (stdio):
import asyncio
import json
import sys
import uuid
JSONRPC_VERSION = "2.0"
PROTOCOL_VERSION = "2025-03-26"
# --- JSON-RPC Message Sending ---
async def send_message(msg_dict):
"""Encodes and sends a JSON-RPC message to stdout."""
message = json.dumps(msg_dict) + '\n'
sys.stdout.write(message)
sys.stdout.flush() # Important for stdio
# Consider adding logging for sent messages here
# sys.stderr.write(f"Sent: {message}")
# sys.stderr.flush()
async def send_response(req_id, result_payload):
await send_message({
"jsonrpc": JSONRPC_VERSION,
"id": req_id,
"result": result_payload,
})
async def send_error(req_id, code, message, data=None):
error_obj = {"code": code, "message": message}
if data:
error_obj["data"] = data
await send_message({
"jsonrpc": JSONRPC_VERSION,
"id": req_id,
"error": error_obj,
})
async def send_notification(method, params=None):
msg = {"jsonrpc": JSONRPC_VERSION, "method": method}
if params:
msg["params"] = params
await send_message(msg)
# --- Request Handlers ---
async def handle_initialize(req_id, params):
client_version = params.get("protocolVersion")
# Basic version check (adapt as needed)
if client_version != PROTOCOL_VERSION:
# Example: Respond with supported version if different
# await send_error(req_id, -32602, "Unsupported protocol version", {"supported": [PROTOCOL_VERSION]})
# Or just accept if compatible range
pass # Assuming compatibility for this example
server_capabilities = {
# Declare supported features, e.g., only ping
# "resources": {}, "prompts": {}, "tools": {}
}
server_info = {"name": "MyMinimalMCPPythonServer", "version": "0.1.0"}
await send_response(req_id, {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": server_capabilities,
"serverInfo": server_info,
})
async def handle_initialized(params):
# Server is now operational, can start sending its own requests/notifications
sys.stderr.write("Server initialized by client.\n")
sys.stderr.flush()
# Example: Maybe send a welcome log message if logging capability enabled
# if "logging" in negotiated_server_capabilities:
# await send_notification("notifications/message", {"level": "info", "data": "Server ready."})
async def handle_ping(req_id, params):
await send_response(req_id, {}) # Empty result signifies success
async def handle_unknown_method(req_id, method):
await send_error(req_id, -32601, f"Method not found: {method}")
# --- Main Loop ---
async def main():
reader = asyncio.StreamReader()
protocol = asyncio.StreamReaderProtocol(reader)
await asyncio.get_event_loop().connect_read_pipe(lambda: protocol, sys.stdin)
while not reader.at_eof():
line_bytes = await reader.readline()
if not line_bytes:
break
line = line_bytes.decode('utf-8').strip()
if not line:
continue
# sys.stderr.write(f"Received: {line}\n") # Debugging
# sys.stderr.flush()
try:
message = json.loads(line)
# Basic validation (add more robust checks)
if not isinstance(message, dict) or message.get("jsonrpc") != JSONRPC_VERSION:
# Handle batch requests if needed here
if isinstance(message, list):
# Basic batch handling stub - respond individually or batch response
# For simplicity, ignoring batch specifics here
sys.stderr.write("Batch request received - minimal example ignores\n")
sys.stderr.flush()
continue
# Send error if possible (no ID for notification error)
req_id = message.get("id") if isinstance(message, dict) else None
if req_id is not None:
await send_error(req_id, -32600, "Invalid Request")
continue
req_id = message.get("id")
method = message.get("method")
params = message.get("params", {})
if method:
if req_id is not None: # It's a Request
if method == "initialize":
await handle_initialize(req_id, params)
elif method == "ping":
await handle_ping(req_id, params)
# Add handlers for other supported methods (resources/list, tools/call etc.)
# elif method == "resources/list":
# await handle_resources_list(req_id, params)
else:
await handle_unknown_method(req_id, method)
else: # It's a Notification
if method == "notifications/initialized":
await handle_initialized(params)
elif method == "notifications/cancelled":
# Handle cancellation logic
pass
elif method == "notifications/progress":
# Handle progress updates if applicable
pass
# Ignore unknown notifications
else:
# It might be a response to a server-sent request (if server sends requests)
# Or an invalid message
if req_id is not None and ("result" in message or "error" in message):
# Handle response if server expects one
pass
elif req_id is not None: # Invalid message with ID but no method/result/error
await send_error(req_id, -32600, "Invalid Request structure")
except json.JSONDecodeError:
# Cannot determine ID, send general error if possible? Difficult.
sys.stderr.write(f"Failed to decode JSON: {line}\n")
sys.stderr.flush()
# Maybe send error with null ID if spec allows? Check JSON-RPC spec.
# For now, just log error.
except Exception as e:
# Catch-all for handler errors
sys.stderr.write(f"Error processing message: {e}\n")
sys.stderr.flush()
req_id = message.get("id") if 'message' in locals() and isinstance(message, dict) else None
if req_id is not None:
await send_error(req_id, -32603, "Internal server error")
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
sys.stderr.write("Server shutting down.\n")
sys.stderr.flush()
Further Resources:
- Specification Repo: github.com/modelcontextprotocol/specification (Includes JSON Schema)
- Python SDK: github.com/modelcontextprotocol/python-sdk
- Website: modelcontextprotocol.io