Dev.to AI 🤖 Ai Tue, 03 Nov 2026 03:38:19 +0000 👁 0 📖 3 min read

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

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 envir

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\": \"[email protected]\", \"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

📰 Read the original article on Dev.to AI

Originally published by Dev.to AI. Aggregated on AIWithGhost for educational purposes — full credit and traffic to the original publisher.

← News 🤖 Ai 🏠 Home