Build an MCP Server from Scratch
Implement the Model Context Protocol by hand: JSON-RPC messages, a tool registry that writes its own JSON Schema, and the dispatcher that lets an AI assistant like Claude call your Python functions.
How it works in the real world
The Model Context Protocol (MCP) is the open standard AI apps such as Claude use to call tools that other people write. An MCP server is a program that says "here are my tools, and here is how to call them"; the AI app is the client.
- Transport: client and server exchange JSON-RPC 2.0 messages, one JSON object per line over stdin/stdout (or over HTTP).
- Handshake: the client sends
initialize; the server answers with its protocol version, capabilities and name. - Discovery:
tools/listreturns every tool with a JSON Schema describing its arguments, so the model knows what it may send. - Execution:
tools/callruns a tool and returns content blocks; failures come back as errors the model can read and recover from.
Libraries such as the official mcp SDK hide all of this behind one decorator. Building it yourself first means you will know exactly what that decorator is doing.
Speak JSON-RPC
Every MCP message is a JSON-RPC 2.0 object. Finish classify(line) so it returns request #ID METHOD when there is an id, notification METHOD when there is not, or invalid: parse error / invalid: wrong version / invalid: missing method.
Wrap json.loads in try/except json.JSONDecodeError. Check msg.get("jsonrpc") == "2.0", then isinstance(msg.get("method"), str), then whether "id" in msg.
import json
raw = [
'{"jsonrpc": "2.0", "id": 1, "method": "initialize", "params": {"protocolVersion": "2025-06-18"}}',
'{"jsonrpc": "2.0", "method": "notifications/initialized"}',
'{"jsonrpc": "2.0", "id": 2, "method": "tools/list"}',
'{"jsonrpc": "1.0", "id": 3, "method": "tools/list"}',
'{"jsonrpc": "2.0", "id": 4}',
"not json at all",
]
def classify(line):
try:
msg = json.loads(line)
except json.JSONDecodeError:
return "invalid: parse error"
if msg.get("jsonrpc") != "2.0":
return "invalid: wrong version"
if not isinstance(msg.get("method"), str):
return "invalid: missing method"
if "id" in msg:
return f"request #{msg['id']} {msg['method']}"
return f"notification {msg['method']}"
for line in raw:
print(classify(line))
Run your code to check this step…
A tool registry that writes its own schema
The model can only call a tool if it knows the arguments. Finish the @tool decorator: read the function's signature with inspect, map int/float/str/bool to JSON Schema types, mark parameters without defaults as required, and store the definition in TOOLS. Then print each definition as JSON.
for name, param in inspect.signature(fn).parameters.items(): use TYPE_NAMES[param.annotation], and param.default is inspect.Parameter.empty means the argument is required. inspect.getdoc(fn) gives the docstring.
import inspect
import json
TOOLS = {}
TYPE_NAMES = {int: "integer", float: "number", str: "string", bool: "boolean"}
def tool(fn):
"""Register fn as an MCP tool, deriving its JSON Schema from the signature."""
props, required = {}, []
for name, param in inspect.signature(fn).parameters.items():
props[name] = {"type": TYPE_NAMES[param.annotation]}
if param.default is inspect.Parameter.empty:
required.append(name)
TOOLS[fn.__name__] = {
"fn": fn,
"definition": {
"name": fn.__name__,
"description": inspect.getdoc(fn),
"inputSchema": {"type": "object", "properties": props, "required": required},
},
}
return fn
@tool
def add(a: int, b: int):
"""Add two integers."""
return a + b
@tool
def word_count(text: str):
"""Count the words in a piece of text."""
return len(text.split())
@tool
def convert_temp(celsius: float, unit: str = "F"):
"""Convert a Celsius temperature to F or K."""
if unit == "F":
return round(celsius * 9 / 5 + 32, 1)
if unit == "K":
return round(celsius + 273.15, 2)
raise ValueError(f"unknown unit {unit}")
for entry in TOOLS.values():
print(json.dumps(entry["definition"]))
Run your code to check this step…
The dispatcher: initialize, list, call
Write handle(msg) to answer three methods: initialize (protocol version 2025-06-18, {"tools": {}} capability, server name lwp-demo version 1.0.0), tools/list and tools/call, which returns the value as a text content block with isError: false.
For tools/call: TOOLS[params["name"]]["fn"](**params.get("arguments", {})), then wrap str(value) in {"content": [{"type": "text", "text": ...}], "isError": False}.
import inspect
import json
TOOLS = {}
TYPE_NAMES = {int: "integer", float: "number", str: "string", bool: "boolean"}
def tool(fn):
"""Register fn as an MCP tool, deriving its JSON Schema from the signature."""
props, required = {}, []
for name, param in inspect.signature(fn).parameters.items():
props[name] = {"type": TYPE_NAMES[param.annotation]}
if param.default is inspect.Parameter.empty:
required.append(name)
TOOLS[fn.__name__] = {
"fn": fn,
"definition": {
"name": fn.__name__,
"description": inspect.getdoc(fn),
"inputSchema": {"type": "object", "properties": props, "required": required},
},
}
return fn
@tool
def add(a: int, b: int):
"""Add two integers."""
return a + b
@tool
def word_count(text: str):
"""Count the words in a piece of text."""
return len(text.split())
@tool
def convert_temp(celsius: float, unit: str = "F"):
"""Convert a Celsius temperature to F or K."""
if unit == "F":
return round(celsius * 9 / 5 + 32, 1)
if unit == "K":
return round(celsius + 273.15, 2)
raise ValueError(f"unknown unit {unit}")
def ok(req_id, result):
return {"jsonrpc": "2.0", "id": req_id, "result": result}
def handle(msg):
method, params = msg["method"], msg.get("params", {})
if method == "initialize":
return ok(msg["id"], {
"protocolVersion": "2025-06-18",
"capabilities": {"tools": {}},
"serverInfo": {"name": "lwp-demo", "version": "1.0.0"},
})
if method == "tools/list":
return ok(msg["id"], {"tools": [t["definition"] for t in TOOLS.values()]})
if method == "tools/call":
value = TOOLS[params["name"]]["fn"](**params.get("arguments", {}))
return ok(msg["id"], {"content": [{"type": "text", "text": str(value)}], "isError": False})
print(json.dumps(handle({"jsonrpc": "2.0", "id": 1, "method": "initialize"})))
listed = handle({"jsonrpc": "2.0", "id": 2, "method": "tools/list"})
print([t["name"] for t in listed["result"]["tools"]])
print(json.dumps(handle({"jsonrpc": "2.0", "id": 3, "method": "tools/call",
"params": {"name": "convert_temp", "arguments": {"celsius": 21.5}}})))
Run your code to check this step…
Fail like a real server
Models make mistakes, so the server must answer them clearly. Wrap handle in serve(msg): notifications get no reply (None); an unknown method is error -32601; an unknown tool or wrong arguments is -32602; and a tool that raises becomes a normal result with isError: true so the model can read the message and try again.
Check "id" not in msg first. Then test the method name and params["name"] in TOOLS before calling. Calling fn(**arguments) with a missing argument raises TypeError (bad arguments); any other exception from the tool becomes {"content": [{"type": "text", "text": f"Error: {exc}"}], "isError": True}.
import inspect
import json
TOOLS = {}
TYPE_NAMES = {int: "integer", float: "number", str: "string", bool: "boolean"}
def tool(fn):
"""Register fn as an MCP tool, deriving its JSON Schema from the signature."""
props, required = {}, []
for name, param in inspect.signature(fn).parameters.items():
props[name] = {"type": TYPE_NAMES[param.annotation]}
if param.default is inspect.Parameter.empty:
required.append(name)
TOOLS[fn.__name__] = {
"fn": fn,
"definition": {
"name": fn.__name__,
"description": inspect.getdoc(fn),
"inputSchema": {"type": "object", "properties": props, "required": required},
},
}
return fn
@tool
def add(a: int, b: int):
"""Add two integers."""
return a + b
@tool
def word_count(text: str):
"""Count the words in a piece of text."""
return len(text.split())
@tool
def convert_temp(celsius: float, unit: str = "F"):
"""Convert a Celsius temperature to F or K."""
if unit == "F":
return round(celsius * 9 / 5 + 32, 1)
if unit == "K":
return round(celsius + 273.15, 2)
raise ValueError(f"unknown unit {unit}")
def ok(req_id, result):
return {"jsonrpc": "2.0", "id": req_id, "result": result}
def handle(msg):
method, params = msg["method"], msg.get("params", {})
if method == "initialize":
return ok(msg["id"], {
"protocolVersion": "2025-06-18",
"capabilities": {"tools": {}},
"serverInfo": {"name": "lwp-demo", "version": "1.0.0"},
})
if method == "tools/list":
return ok(msg["id"], {"tools": [t["definition"] for t in TOOLS.values()]})
if method == "tools/call":
value = TOOLS[params["name"]]["fn"](**params.get("arguments", {}))
return ok(msg["id"], {"content": [{"type": "text", "text": str(value)}], "isError": False})
def err(req_id, code, message):
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}
def serve(msg):
if "id" not in msg:
return None
method, params = msg["method"], msg.get("params", {})
if method not in ("initialize", "tools/list", "tools/call"):
return err(msg["id"], -32601, f"Method not found: {method}")
if method == "tools/call":
if params.get("name") not in TOOLS:
return err(msg["id"], -32602, f"Unknown tool: {params.get('name')}")
try:
inspect.signature(TOOLS[params["name"]]["fn"]).bind(**params.get("arguments", {}))
except TypeError as exc:
return err(msg["id"], -32602, f"Invalid arguments: {exc}")
try:
return handle(msg)
except Exception as exc:
return ok(msg["id"], {"content": [{"type": "text", "text": f"Error: {exc}"}], "isError": True})
return handle(msg)
session = [
{"jsonrpc": "2.0", "method": "notifications/initialized"},
{"jsonrpc": "2.0", "id": 5, "method": "resources/list"},
{"jsonrpc": "2.0", "id": 6, "method": "tools/call", "params": {"name": "delete_everything", "arguments": {}}},
{"jsonrpc": "2.0", "id": 7, "method": "tools/call", "params": {"name": "add", "arguments": {"a": 1}}},
{"jsonrpc": "2.0", "id": 8, "method": "tools/call", "params": {"name": "convert_temp", "arguments": {"celsius": 10, "unit": "X"}}},
{"jsonrpc": "2.0", "id": 9, "method": "tools/call", "params": {"name": "add", "arguments": {"a": 19, "b": 23}}},
]
for msg in session:
reply = serve(msg)
print("(no reply)" if reply is None else json.dumps(reply))
Run your code to check this step…
Project shipped!
You just built a complete, working build an mcp server from scratch: the same architecture running in production software. Pick your next build →
Take it further
You have written the protocol by hand. Now run the real thing on your own machine and plug it into an AI assistant. Nothing below is graded: it runs on your own machine, not in this tab.
Stretch goals
- Add a
tools/listpagination cursor so a server with hundreds of tools does not send them all at once. - Read newline-delimited JSON from
sys.stdinand write replies tosys.stdout: that is the whole stdio transport, and your step 4 code becomes a working server. - Add a tool that reads a CSV from disk and returns summary statistics, then ask an assistant questions about your own data.
- Write tests with
pytestthat send malformed messages and assert on every error code.
server.py: the same server with the official MCP Python SDK
# pip install "mcp[cli]"
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("lwp-demo")
@mcp.tool()
def add(a: int, b: int) -> int:
"""Add two integers."""
return a + b
@mcp.tool()
def convert_temp(celsius: float, unit: str = "F") -> float:
"""Convert a Celsius temperature to F or K."""
if unit == "F":
return round(celsius * 9 / 5 + 32, 1)
if unit == "K":
return round(celsius + 273.15, 2)
raise ValueError(f"unknown unit {unit}")
if __name__ == "__main__":
mcp.run() # stdio transport
# Try it:
# npx @modelcontextprotocol/inspector python server.py
# Connect it to Claude Code:
# claude mcp add lwp-demo -- python /full/path/to/server.py
Ask an AI to review your build
I built a Model Context Protocol server in Python by hand (JSON-RPC 2.0 dispatcher, a tool registry that derives JSON Schema from type hints, and error handling for -32601 and -32602). Here is my code: [paste it]. Review it like a senior engineer: point out protocol mistakes, missing edge cases (notifications, bad arguments, tools that raise), and one thing I should test next. Do not rewrite it for me; explain what to change and why.