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

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.

---

1. The Anatomy of an MCP Message (JSON-RPC 2.0)

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.

The Tool Invocation Request

{
  "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;"
    }
  }
}

The Tool Execution Response

{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "content": [
      {
        "type": "text",
        "text": "[{\"id\": 101, \"email\": \"[e-mail]\", \"status\": \"active\"}]"
      }
    ],
    "isError": false
  }
}

---

2. Core Protocol Models with Pydantic V2

To maintain 100% type safety and automatic schema validation, we define our core protocol data contracts:

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

---

3. Dynamic Tool Schema Extraction from Type Hints

Rather than maintaining duplicate JSON schemas by hand, our server inspects Python functions, extracts docstrings, and derives parameters directly from native type hints:

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

---

4. The Asynchronous Stdio Transport

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:

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.

---

5. Integrating with Claude Desktop & Cursor

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.

---

📚 Master the Full Architecture

Want the complete, production-ready implementation?
Our new flagship guide and commercial repository is now live:

🚀 Production Model Context Protocol (MCP) with Python
Includes:

• Full standards-compliant EPUB 3.0 & 6x9 Print PDF
• Standalone Dark-Mode Offline Web Reader
• Complete mcpmaster_pro Python Repository (Server, Client, SQLite Gateway & Test Suite)
• Use coupon code LAUNCH20 for 20% off storewide!

Or get the complete 4-book + 8-repository bundle:
🏛️ The Sovereign Software Architect Quad-Vault