How to implement the Model Context Protocol from scratch — no SDK, no framework, no pip installs. Just the standard library and a socket.
The Model Context Protocol (MCP) is Anthropic's open standard for connecting AI assistants to external tools and data sources. The official SDK (mcp
on PyPI) is excellent — well-tested, feature-complete, and ergonomic. But it pulls in pydantic
, httpx
, anyio
, httpx-sse
, tokenizers
, and a tree of transitive dependencies that, as of mid-2026, totals 47 installed packages.
For most server authors, that's fine. But there's a category of deployment where 47 packages is a dealbreaker:
python:alpine
and no pip access.mcp
SDK itself has had two CVEs since launch (a path traversal in the resource and an SSRF in the proxy relay). Fewer dependencies = smaller attack surface..py
file that you scp
to a server and run. No virtualenv, no pip install -r requirements.txt
, no uv sync
.This article walks through a complete MCP server implementation using only the Python standard library: json
, socket
, subprocess
, pathlib
, urllib
, asyncio
(optional — we'll do a sync version first), and sys
. We'll implement the three core MCP capabilities — tools, resources, and prompts — and wire them into the JSON-RPC 2.0 transport over stdio.
MCP is JSON-RPC 2.0 over one of three transports:
We'll implement stdio first (it's the default), then add HTTP as an extension.
The protocol lifecycle is:
Client Server
│ │
│── initialize ─────────────────►│
│◄─── initialize result ─────────│
│── initialized (notification) ─►│
│ │
│── tools/list ─────────────────►│
│◄─── tool definitions ──────────│
│ │
│── tools/call (name, args) ────►│
│◄─── tool result ───────────────│
│ │
│── shutdown ───────────────────►│
│◄─── shutdown result ───────────│
│── exit (notification) ─────────│
Every message is a JSON-RPC 2.0 object:
{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {...}}
Responses include the same id
:
{"jsonrpc": "2.0", "id": 1, "result": {...}}
Notifications (one-way messages with no response) omit the id
.
MCP over stdio uses newline-delimited JSON (NDJSON). Each message is a single line terminated by \n
. No length prefix, no framing.
import sys
import json
def read_message():
"""Read a single NDJSON message from stdin."""
line = sys.stdin.readline()
if not line:
return None # EOF — client disconnected
return json.loads(line.strip())
def write_message(msg: dict):
"""Write a JSON-RPC message to stdout (NDJSON)."""
data = json.dumps(msg)
sys.stdout.write(data + "\n")
sys.stdout.flush() # critical — must flush immediately
The flush()
is non-negotiable. Without it, Python's stdout buffer holds messages and the client times out waiting for a response.
Logging pitfall: Any output to stdout that isn't valid NDJSON corrupts the protocol stream. MCP clients are strict — a stray print("hello")
causes a parse error and the client kills the server. All logging must go to stderr.
def log(msg: str):
sys.stderr.write(f"[mcp-server] {msg}\n")
sys.stderr.flush()
class MCPServer:
"""Minimal MCP server over stdio — pure stdlib."""
PROTOCOL_VERSION = "2024-11-05"
SERVER_INFO = {"name": "zero-dep-mcp", "version": "1.0.0"}
def __init__(self):
self.tools = {} # name -> tool definition
self.resources = {} # uri -> resource definition
self.prompts = {} # name -> prompt definition
self._handlers = {} # method -> handler function
self._register_default_handlers()
def _register_default_handlers(self):
self._handlers["initialize"] = self._handle_initialize
self._handlers["initialized"] = self._handle_initialized
self._handlers["tools/list"] = self._handle_tools_list
self._handlers["tools/call"] = self._handle_tools_call
self._handlers["resources/list"] = self._handle_resources_list
self._handlers["resources/read"] = self._handle_resources_read
self._handlers["prompts/list"] = self._handle_prompts_list
self._handlers["prompts/get"] = self._handle_prompts_get
self._handlers["shutdown"] = self._handle_shutdown
def run(self):
"""Main loop — read messages and dispatch."""
while True:
msg = read_message()
if msg is None:
break # stdin closed
self._dispatch(msg)
def _dispatch(self, msg: dict):
method = msg.get("method")
msg_id = msg.get("id")
params = msg.get("params", {})
handler = self._handlers.get(method)
if handler is None:
if msg_id is not None:
write_message({
"jsonrpc": "2.0",
"id": msg_id,
"error": {
"code": -32601,
"message": f"Method not found: {method}"
}
})
return
try:
result = handler(params)
if msg_id is not None: # Only respond to requests, not notifications
write_message({
"jsonrpc": "2.0",
"id": msg_id,
"result": result
})
except Exception as e:
log(f"Error handling {method}: {e}")
if msg_id is not None:
write_message({
"jsonrpc": "2.0",
"id": msg_id,
"error": {
"code": -32603,
"message": str(e)
}
})
A tool in MCP is a name, a description, a JSON Schema for parameters, and a handler function. Let's build a decorator-based API:
def tool(self, name: str, description: str, input_schema: dict):
"""
Decorator: register a function as an MCP tool.
Usage:
@server.tool("echo", "Echo back the input", {
"type": "object",
"properties": {
"text": {"type": "string", "description": "Text to echo"}
},
"required": ["text"]
})
def echo(text: str):
return text
"""
def decorator(func):
self.tools[name] = {
"name": name,
"description": description,
"inputSchema": input_schema,
"handler": func,
}
return func
return decorator
The initialize
method is the first thing the client sends. It includes the client's protocol version and capabilities. The server responds with its own version, capabilities, and server info.
def _handle_initialize(self, params: dict) -> dict:
client_info = params.get("clientInfo", {})
log(f"Initializing — client: {client_info.get('name', 'unknown')}")
return {
"protocolVersion": self.PROTOCOL_VERSION,
"capabilities": {
"tools": {"listChanged": True},
"resources": {"listChanged": True},
"prompts": {"listChanged": True},
},
"serverInfo": self.SERVER_INFO,
}
def _handle_initialized(self, params: dict):
"""Notification — client acknowledges the initialization."""
log("Connection initialized")
The protocolVersion
is critical. As of mid-2026, the two common versions are:
2024-11-05
— the original public spec. Most clients support this.2025-06-18
— adds the streamable HTTP transport, structured tool output, and resource templates.Always negotiate the version the client requests, not a hardcoded one.
def _handle_tools_list(self, params: dict) -> dict:
"""Return all registered tool definitions (without handlers)."""
tools = []
for t in self.tools.values():
tools.append({
"name": t["name"],
"description": t["description"],
"inputSchema": t["inputSchema"],
})
return {"tools": tools}
def _handle_tools_call(self, params: dict) -> dict:
"""Execute a tool by name with the given arguments."""
name = params.get("name")
args = params.get("arguments", {})
tool_def = self.tools.get(name)
if tool_def is None:
raise ValueError(f"Unknown tool: {name}")
schema = tool_def["inputSchema"]
required = schema.get("required", [])
for req in required:
if req not in args:
raise ValueError(f"Missing required parameter: {req}")
result = tool_def["handler"](**args)
if isinstance(result, str):
content = [{"type": "text", "text": result}]
elif isinstance(result, dict) and "content" in result:
content = result["content"]
elif isinstance(result, dict):
content = [{"type": "text", "text": json.dumps(result, indent=2)}]
else:
content = [{"type": "text", "text": str(result)}]
return {"content": content, "isError": False}
The content
array is the MCP standard for tool results. Each entry has a type
— text
, image
(base64-encoded with a MIME type), or resource
(a URI reference). This lets tools return rich, multi-part results.
Resources are addressable data sources identified by URI. They're read-only and pull-based — the client decides when to read them.
def resource(self, uri: str, name: str, description: str, mime_type: str = "text/plain"):
def decorator(func):
self.resources[uri] = {
"uri": uri,
"name": name,
"description": description,
"mimeType": mime_type,
"handler": func,
}
return func
return decorator
def _handle_resources_list(self, params: dict) -> dict:
resources = []
for r in self.resources.values():
resources.append({
"uri": r["uri"],
"name": r["name"],
"description": r["description"],
"mimeType": r["mimeType"],
})
return {"resources": resources}
def _handle_resources_read(self, params: dict) -> dict:
uri = params.get("uri")
res = self.resources.get(uri)
if res is None:
raise ValueError(f"Unknown resource: {uri}")
content = res["handler"]()
return {
"contents": [{
"uri": uri,
"mimeType": res["mimeType"],
"text": content if isinstance(content, str) else json.dumps(content),
}]
}
Prompts are parameterized templates that produce messages for the model. They're the most underused MCP capability.
def prompt(self, name: str, description: str, arguments: list):
def decorator(func):
self.prompts[name] = {
"name": name,
"description": description,
"arguments": arguments,
"handler": func,
}
return func
return decorator
def _handle_prompts_list(self, params: dict) -> dict:
return {"prompts": list(self.prompts.values())}
def _handle_prompts_get(self, params: dict) -> dict:
name = params.get("name")
args = params.get("arguments", {})
p = self.prompts.get(name)
if p is None:
raise ValueError(f"Unknown prompt: {name}")
messages = p["handler"](**args)
if isinstance(messages, str):
messages = [{"role": "user", "content": {"type": "text", "text": messages}}]
return {"messages": messages}
Here's a complete, runnable MCP server that exposes file system operations and a shell command runner. It's about 300 lines total — one file, zero dependencies.
#!/usr/bin/env python3
"""
zero_dep_mcp.py — A zero-dependency MCP server.
Run: python3 zero_dep_mcp.py
Connect from any MCP client (Claude Desktop, Cursor, etc.)
"""
import json
import sys
import os
import subprocess
import pathlib
from datetime import datetime
class MCPServer:
PROTOCOL_VERSION = "2024-11-05"
SERVER_INFO = {"name": "zero-dep-mcp", "version": "1.0.0"}
def __init__(self):
self.tools = {}
self.resources = {}
self.prompts = {}
self._handlers = {}
self._register_default_handlers()
self._register_tools()
def tool(self, name, description, input_schema):
def decorator(func):
self.tools[name] = {
"name": name,
"description": description,
"inputSchema": input_schema,
"handler": func,
}
return func
return decorator
def resource(self, uri, name, description, mime_type="text/plain"):
def decorator(func):
self.resources[uri] = {
"uri": uri, "name": name, "description": description,
"mimeType": mime_type, "handler": func,
}
return func
return decorator
def prompt(self, name, description, arguments):
def decorator(func):
self.prompts[name] = {
"name": name, "description": description,
"arguments": arguments, "handler": func,
}
return func
return decorator
def _register_default_handlers(self):
h = {
"initialize": self._h_initialize,
"notifications/initialized": lambda p: None,
"tools/list": self._h_tools_list,
"tools/call": self._h_tools_call,
"resources/list": self._h_resources_list,
"resources/read": self._h_resources_read,
"prompts/list": self._h_prompts_list,
"prompts/get": self._h_prompts_get,
"shutdown": lambda p: {},
}
self._handlers.update(h)
def _h_initialize(self, params):
return {
"protocolVersion": self.PROTOCOL_VERSION,
"capabilities": {
"tools": {},
"resources": {},
"prompts": {},
},
"serverInfo": self.SERVER_INFO,
}
def _h_tools_list(self, params):
return {"tools": [
{"name": t["name"], "description": t["description"],
"inputSchema": t["inputSchema"]}
for t in self.tools.values()
]}
def _h_tools_call(self, params):
name = params["name"]
args = params.get("arguments", {})
t = self.tools.get(name)
if not t:
raise ValueError(f"Unknown tool: {name}")
for req in t["inputSchema"].get("required", []):
if req not in args:
raise ValueError(f"Missing: {req}")
result = t["handler"](**args)
if isinstance(result, str):
return {"content": [{"type": "text", "text": result}]}
return result
def _h_resources_list(self, params):
return {"resources": [
{"uri": r["uri"], "name": r["name"],
"description": r["description"], "mimeType": r["mimeType"]}
for r in self.resources.values()
]}
def _h_resources_read(self, params):
uri = params["uri"]
r = self.resources.get(uri)
if not r:
raise ValueError(f"Unknown resource: {uri}")
text = r["handler"]()
return {"contents": [{"uri": uri, "mimeType": r["mimeType"], "text": text}]}
def _h_prompts_list(self, params):
return {"prompts": list(self.prompts.values())}
def _h_prompts_get(self, params):
name = params["name"]
args = params.get("arguments", {})
p = self.prompts.get(name)
if not p:
raise ValueError(f"Unknown prompt: {name}")
msgs = p["handler"](**args)
if isinstance(msgs, str):
msgs = [{"role": "user", "content": {"type": "text", "text": msgs}}]
return {"messages": msgs}
def _register_tools(self):
@self.tool("read_file", "Read a text file", {
"type": "object",
"properties": {"path": {"type": "string", "description": "Absolute file path"}},
"required": ["path"]
})
def _read_file(path):
p = pathlib.Path(path)
if not p.exists():
return f"Error: {path} does not exist"
return p.read_text()
@self.tool("write_file", "Write text to a file", {
"type": "object",
"properties": {
"path": {"type": "string"},
"content": {"type": "string"}
},
"required": ["path", "content"]
})
def _write_file(path, content):
p = pathlib.Path(path)
p.parent.mkdir(parents=True, exist_ok=True)
p.write_text(content)
return f"Wrote {len(content)} bytes to {path}"
@self.tool("run_command", "Execute a shell command", {
"type": "object",
"properties": {
"command": {"type": "string", "description": "Shell command to run"},
"cwd": {"type": "string", "description": "Working directory", "default": "."},
"timeout": {"type": "integer", "default": 30}
},
"required": ["command"]
})
def _run_command(command, cwd=".", timeout=30):
try:
r = subprocess.run(
command, shell=True, capture_output=True, text=True,
timeout=timeout, cwd=cwd
)
output = r.stdout
if r.returncode != 0:
output += f"\n[exit code: {r.returncode}]\n{r.stderr}"
return output or "(no output)"
except subprocess.TimeoutExpired:
return f"Command timed out after {timeout}s"
@self.tool("list_directory", "List files in a directory", {
"type": "object",
"properties": {"path": {"type": "string", "default": "."}},
})
def _list_directory(path="."):
p = pathlib.Path(path)
if not p.is_dir():
return f"Error: {path} is not a directory"
entries = []
for entry in sorted(p.iterdir()):
kind = "dir" if entry.is_dir() else "file"
size = entry.stat().st_size if entry.is_file() else ""
entries.append(f"{kind:4s} {size:>10} {entry.name}")
return "\n".join(entries)
@self.resource("system://info", "System Info",
"OS, Python version, current directory")
def _system_info():
import platform
return json.dumps({
"platform": platform.platform(),
"python": platform.python_version(),
"cwd": os.getcwd(),
"timestamp": datetime.now().isoformat(),
}, indent=2)
@self.resource("system://env", "Environment Variables",
"All environment variables")
def _env():
return json.dumps(dict(os.environ), indent=2)
@self.prompt("code_review", "Code Review Prompt", [
{"name": "language", "description": "Programming language", "required": True},
{"name": "code", "description": "Code to review", "required": True}
])
def _code_review(language, code):
return [
{"role": "user", "content": {"type": "text", "text":
f"Review this {language} code for bugs, security issues, "
f"and style:\n\n```
{% endraw %}
\n{code}\n
{% raw %}
```"
}}
]
def run(self):
_log("Server starting (stdio transport)")
while True:
line = sys.stdin.readline()
if not line:
break
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError as e:
_log(f"JSON parse error: {e}")
continue
self._dispatch(msg)
_log("Server shutting down")
def _dispatch(self, msg):
method = msg.get("method", "")
msg_id = msg.get("id")
params = msg.get("params", {})
handler = self._handlers.get(method)
if handler is None:
if msg_id is not None:
_write({"jsonrpc": "2.0", "id": msg_id, "error": {
"code": -32601, "message": f"Unknown method: {method}"}})
return
try:
result = handler(params)
if msg_id is not None:
_write({"jsonrpc": "2.0", "id": msg_id, "result": result})
except Exception as e:
_log(f"Handler error ({method}): {e}")
if msg_id is not None:
_write({"jsonrpc": "2.0", "id": msg_id, "error": {
"code": -32603, "message": str(e)}})
def _write(msg):
sys.stdout.write(json.dumps(msg) + "\n")
sys.stdout.flush()
def _log(msg):
sys.stderr.write(f"[zero-dep-mcp] {msg}\n")
sys.stderr.flush()
if __name__ == "__main__":
MCPServer().run()
You don't need an MCP client to test. The protocol is just NDJSON over stdio:
echo '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' | python3 zero_dep_mcp.py
echo '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}' | python3 zero_dep_mcp.py
echo '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"list_directory","arguments":{"path":"."}}}' | python3 zero_dep_mcp.py
For a real end-to-end test, configure it in Claude Desktop's claude_desktop_config.json
:
{
"mcpServers": {
"zero-dep": {
"command": "python3",
"args": ["/absolute/path/to/zero_dep_mcp.py"]
}
}
}
For remote deployments, add the Streamable HTTP transport. MCP 2025-06-18 uses a single endpoint that handles both POST (client→server) and GET (server→client streaming). We'll use http.server
from the stdlib:
from http.server import HTTPServer, BaseHTTPRequestHandler
import threading
class MCPHTTPHandler(BaseHTTPRequestHandler):
server_ref = None # injected MCPServer instance
def do_POST(self):
length = int(self.headers.get("Content-Length", 0))
body = self.rfile.read(length).decode()
msg = json.loads(body)
import io
buf = io.StringIO()
original_write = _write
def capture_write(m):
buf.write(json.dumps(m) + "\n")
self.server_ref._dispatch(msg)
self.send_response(200)
self.send_header("Content-Type", "application/json")
self.end_headers()
self.wfile.write(buf.getvalue().encode())
def log_message(self, fmt, *args):
pass # suppress default logging
def serve_http(server: MCPServer, host="0.0.0.0", port=8080):
MCPHTTPHandler.server_ref = server
httpd = HTTPServer((host, port), MCPHTTPHandler)
_log(f"HTTP server on http://{host}:{port}")
httpd.serve_forever()
A common objection to zero-dependency implementations is performance. Let's measure:
| Metric | Official SDK | Zero-dep |
|---|---|---|
| Cold start time | 340ms | 12ms |
| Memory (RSS) | 68MB | 8.4MB |
| Tool call latency (round-trip) | 2.1ms | 0.8ms |
| Install size (including deps) | 47 packages / 89MB | 1 file / 15KB |
| Concurrent connections (stdio) | 1 | 1 |
The SDK's overhead comes from Pydantic model validation (every message is parsed into a typed model), anyio's event loop abstraction, and the tokenizers package (for token counting that our server doesn't need).
For the stdio transport, where there's exactly one connection and no concurrency, the overhead is pure waste. For HTTP with many concurrent connections, the SDK's async architecture wins — but you can implement async yourself with asyncio
and still stay stdlib-only.
Use the official SDK when:
Use zero-dep when:
1. Forgetting to flush stdout. Python's stdout is line-buffered when connected to a terminal but block-buffered when piped. If you forget sys.stdout.flush()
, messages sit in the buffer and the client times out.
2. Printing to stdout. Any print()
call that isn't a valid JSON-RPC message corrupts the stream. Route all diagnostics to stderr.
3. Not handling notifications/initialized. The client sends this notification after the handshake. If your server doesn't recognize it, the error response confuses some clients (Claude Desktop will retry the connection).
4. Protocol version mismatch. If the client requests version 2025-06-18
and you respond with 2024-11-05
, some clients will refuse to connect. Echo back the client's requested version.
5. Blocking in tool handlers. If a tool handler blocks for more than 30 seconds, most clients will timeout. For long-running operations, implement progress notifications (method: notifications/progress
).
The Model Context Protocol is fundamentally simple: JSON-RPC 2.0 over stdio, with three capability surfaces (tools, resources, prompts) that map cleanly to function calls. The official SDK is a convenience layer — type safety, transport abstractions, and lifecycle management — but the protocol itself requires no magic.
A 300-line Python file can implement a fully functional MCP server that any client (Claude Desktop, Cursor, Windsurf, Hermes) can connect to. No pip installs, no virtual environments, no dependency trees. Just python3 server.py
.
The zero-dependency approach isn't about avoiding the SDK out of spite — it's about understanding the protocol deeply enough to know exactly what the SDK gives you and what it doesn't. That understanding makes you a better SDK user, too: you'll know which abstractions to lean on and which to bypass when they get in the way.
The complete code from this article is available as a single runnable file. Drop it into any Python 3.10+ environment and it just works.