Building Production Model Context Protocol (MCP) Servers in Pure Python: Stdio, SSE & Pydantic V2

A developer published an architectural guide for building production-grade Model Context Protocol (MCP) servers in pure Python using asyncio, json, and Pydantic V2. The guide covers JSON-RPC 2.0 wire format, Pydantic data contracts for type-safe schema validation, automatic tool-schema extraction from Python type hints and docstrings, and stdio-based transport for clients like Claude Desktop and Cursor.

In the rapidly evolving landscape of agentic AI, connecting Large Language Models to real-world data and execution environments has historically been plagued by fragmentation. Every developer was writing bespoke function-calling wrappers, brittle regex parsers, and custom HTTP endpoints. The Model Context Protocol MCP , open-sourced by Anthropic and rapidly adopted by Claude Desktop , Cursor , and Antigravity , has emerged as the universal standard—the USB-C of Artificial Intelligence . In this architectural guide, we build a production-grade, asynchronous MCP server from scratch in pure Python using asyncio , json , and pydantic . At the wire level, MCP communicates using pure JSON-RPC 2.0 . Every message transmitted between client and server is a newline-delimited JSON payload. { "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "execute sqlite query", "arguments": { "db path": "production.db", "query": "SELECT id, email, status FROM customers WHERE active = 1;" } } } { "jsonrpc": "2.0", "id": 1, "result": { "content": { "type": "text", "text": " {\"id\": 101, \"email\": \"alex@example.com\", \"status\": \"active\"} " } , "isError": false } } To maintain 100% type safety and automatic schema validation, we define our core protocol data contracts: python from typing import Any, Dict, List, Optional, Union from pydantic import BaseModel, Field class JSONRPCRequest BaseModel : jsonrpc: str = "2.0" id: Optional Union str, int = None method: str params: Optional Dict str, Any = None class JSONRPCResponse BaseModel : jsonrpc: str = "2.0" id: Optional Union str, int = None result: Optional Any = None error: Optional Dict str, Any = None class ToolInputSchema BaseModel : type: str = "object" properties: Dict str, Any = Field default factory=dict required: List str = Field default factory=list class ToolDefinition BaseModel : name: str description: str inputSchema: ToolInputSchema class CallToolResult BaseModel : content: List Dict str, str = Field default factory=list isError: bool = False Rather than maintaining duplicate JSON schemas by hand, our server inspects Python functions, extracts docstrings, and derives parameters directly from native type hints: python import inspect from typing import Callable, get type hints def extract schema func: Callable - ToolDefinition: sig = inspect.signature func type hints = get type hints func properties = {} required = type map = { int: "integer", float: "number", str: "string", bool: "boolean", list: "array", dict: "object" } for param name, param in sig.parameters.items : if param name in "self", "cls" : continue py type = type hints.get param name, str json type = type map.get py type, "string" properties param name = {"type": json type, "description": f"Parameter {param name}"} if param.default is inspect.Parameter.empty: required.append param name return ToolDefinition name=func. name , description= func. doc or "" .strip , inputSchema=ToolInputSchema properties=properties, required=required For local tools configured in Claude Desktop or Cursor, communication happens over standard I/O streams sys.stdin and sys.stdout . Stream flushing is critical: python import sys import json import asyncio async def run stdio server server : loop = asyncio.get event loop reader = asyncio.StreamReader protocol = asyncio.StreamReaderProtocol reader await loop.connect read pipe lambda: protocol, sys.stdin while True: line = await reader.readline if not line: break try: req json = json.loads line.decode "utf-8" .strip resp = await server.handle request req json if resp is not None: sys.stdout.write json.dumps resp + "\n" sys.stdout.flush except Exception as e: err resp = {"jsonrpc": "2.0", "id": None, "error": {"code": -32700, "message": str e }} sys.stdout.write json.dumps err resp + "\n" sys.stdout.flush Warning: Never use unbuffered print statements inside an MCP Stdio server. Any rogue stdout logging will break the JSON-RPC stream parser. Always direct internal server logs to sys.stderr . Once your Python script is ready, add it to your claude desktop config.json : { "mcpServers": { "my python tools": { "command": "python", "args": "-m", "mcpmaster pro.server cli" } } } Now, Claude Desktop natively sees your custom tools, knows their exact parameter requirements, and executes them with full local privacy and $0 ongoing cloud API subscription bills. Want the complete, production-ready implementation? Our new flagship guide and commercial repository is now live: 🚀 Production Model Context Protocol MCP with Python https://antigone16.gumroad.com/l/production-mcp-python Includes: mcpmaster pro LAUNCH20 Or get the complete 4-book + 8-repository bundle: 🏛️ The Sovereign Software Architect Quad-Vault https://antigone16.gumroad.com/l/sovereign-architect-quad-vault