2. The protocol from scratch¶
Intermediate · 15 min read
SDKs hide the protocol, which is fine until something breaks and you need to read the JSON. This page builds a small MCP server and client with only the standard library, so you can see every message. It follows the stateless 2026-07-28 spec from page 1. On the next page you'll swap all of it for the official SDK in about 20 lines.
A teaching server, not a compliant one
The real spec has more required fields (cache hints, pagination, many content types) and more edge cases. Use this to learn the protocol. Use the SDK for anything real.
2.1 The server's data: tools, resources, prompts¶
A server is a few registries plus a dispatcher. Start with the data:
import json
PROTOCOL_VERSION = "2026-07-28"
SERVER_INFO = {"name": "orders", "version": "0.1.0"}
ORDERS = {"A100": {"status": "shipped", "eta": "Friday", "total_inr": 2499},
"A101": {"status": "processing", "eta": "Monday", "total_inr": 899}}
TOOLS = {
"get_order_status": {
"description": "Shipping status of an order. Use when the user gives an order ID like A100.",
"inputSchema": {"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"]},
},
"cancel_order": {
"description": "Cancel an order. The user is asked to confirm first.",
"inputSchema": {"type": "object",
"properties": {"order_id": {"type": "string"}},
"required": ["order_id"]},
},
}
RESOURCES = {"policy://refunds": "Refunds within 7 days of delivery. Damaged items: full refund."}
print(f"{len(TOOLS)} tools, {len(RESOURCES)} resource")
2.2 Tool code and argument checks¶
The model's arguments are untrusted input, so check them against the schema before running anything. A failed
check, or a failure inside the tool, becomes a result with isError: true that the model can read and react to:
class ToolFailure(Exception):
"""A failure the model should see and recover from."""
def check_args(schema: dict, args: dict) -> None:
missing = [k for k in schema["required"] if k not in args]
extra = [k for k in args if k not in schema["properties"]]
wrong = [k for k, v in args.items() if k in schema["properties"] and not isinstance(v, str)]
problems = [f"missing {k!r}" for k in missing] + [f"unexpected {k!r}" for k in extra]
problems += [f"{k!r} must be a string" for k in wrong]
if problems:
raise ToolFailure("; ".join(problems))
def get_order_status(order_id: str) -> dict:
if order_id not in ORDERS:
raise ToolFailure(f"order {order_id} not found: IDs look like A100")
order = ORDERS[order_id]
return {"order_id": order_id, "status": order["status"], "eta": order["eta"]}
def text_result(text: str, structured=None, is_error=False) -> dict:
result = {"content": [{"type": "text", "text": text}], "isError": is_error, "resultType": "complete"}
if structured is not None:
result["structuredContent"] = structured
return result
print(text_result("hello"))
2.3 Multi round-trip: asking the user mid-call¶
cancel_order shouldn't run without a "yes" from the user. In the stateless spec the server can't send its own
request in the middle of a call. Instead it returns early with resultType: "input_required" and a list of
questions. The client asks the user and retries the same call with the answers in inputResponses.
sequenceDiagram
participant C as Client (host)
participant S as Server
C->>S: tools/call cancel_order {order_id: A100}
S-->>C: resultType: input_required<br/>inputRequests: {confirm: elicitation "Really cancel A100?"}
Note over C: host shows the question to the user
C->>S: tools/call cancel_order {order_id: A100}<br/>inputResponses: {confirm: {action: accept, content: {confirm: true}}}
S-->>C: resultType: complete — "A100 cancelled"def cancel_order(order_id: str, input_responses: dict | None) -> dict:
answer = (input_responses or {}).get("confirm")
if answer is None: # first round: ask
return {
"resultType": "input_required",
"inputRequests": {"confirm": {
"method": "elicitation/create",
"params": {"mode": "form", "message": f"Really cancel order {order_id}?",
"requestedSchema": {"type": "object",
"properties": {"confirm": {"type": "boolean"}},
"required": ["confirm"]}}}},
}
if answer.get("action") == "accept" and answer.get("content", {}).get("confirm"):
ORDERS[order_id]["status"] = "cancelled"
return text_result(f"{order_id} cancelled")
return text_result(f"{order_id} kept: the user did not confirm")
The server keeps no memory between the two rounds. Everything it needs comes back in the retry. That's what lets any replica behind a load balancer answer the second request.
2.4 The dispatcher¶
Now the JSON-RPC layer: read a request, check the protocol version, route by method, and wrap the answer as a
response or an error.
PARSE_ERROR, INVALID_REQUEST, METHOD_NOT_FOUND, INVALID_PARAMS = -32700, -32600, -32601, -32602
UNSUPPORTED_PROTOCOL_VERSION = -32022
def handle(raw: str) -> str | None:
try:
msg = json.loads(raw)
except json.JSONDecodeError:
return json.dumps({"jsonrpc": "2.0", "id": None, "error": {"code": PARSE_ERROR, "message": "Parse error"}})
if "id" not in msg:
return None # a notification: no reply
reply = {"jsonrpc": "2.0", "id": msg["id"]}
try:
reply["result"] = route(msg.get("method"), msg.get("params") or {})
reply["result"].setdefault("resultType", "complete")
reply["result"]["_meta"] = {"io.modelcontextprotocol/serverInfo": SERVER_INFO}
except RpcError as err:
reply["error"] = {"code": err.code, "message": str(err), **({"data": err.data} if err.data else {})}
return json.dumps(reply)
class RpcError(Exception):
def __init__(self, code, message, data=None):
super().__init__(message)
self.code, self.data = code, data
def route(method: str, params: dict) -> dict:
meta = params.get("_meta", {})
if method == "server/discover":
return {"supportedVersions": [PROTOCOL_VERSION],
"capabilities": {"tools": {}, "resources": {}},
"serverInfo": SERVER_INFO}
version = meta.get("io.modelcontextprotocol/protocolVersion")
if version != PROTOCOL_VERSION:
raise RpcError(UNSUPPORTED_PROTOCOL_VERSION, f"unsupported protocol version {version!r}",
{"supported": [PROTOCOL_VERSION]})
if method == "tools/list":
return {"tools": [{"name": n, **spec} for n, spec in TOOLS.items()]}
if method == "tools/call":
return call_tool(params)
if method == "resources/list":
return {"resources": [{"uri": u, "name": u.split("//")[1], "mimeType": "text/plain"} for u in RESOURCES]}
if method == "resources/read":
uri = params.get("uri")
if uri not in RESOURCES:
raise RpcError(INVALID_PARAMS, f"resource not found: {uri}")
return {"contents": [{"uri": uri, "mimeType": "text/plain", "text": RESOURCES[uri]}]}
raise RpcError(METHOD_NOT_FOUND, f"method not found: {method}")
def call_tool(params: dict) -> dict:
name, args = params.get("name"), params.get("arguments") or {}
if name not in TOOLS:
raise RpcError(INVALID_PARAMS, f"unknown tool: {name}")
try:
check_args(TOOLS[name]["inputSchema"], args)
if name == "get_order_status":
data = get_order_status(**args)
return text_result(json.dumps(data), structured=data)
return cancel_order(args["order_id"], params.get("inputResponses"))
except ToolFailure as err:
return text_result(f"Error: {err}", is_error=True)
Note the two kinds of failure. An unknown tool or unknown method is a protocol error (error with a code).
A tool that ran and failed returns a normal result with isError: true, because the model should see it.
2.5 A client¶
The client side is small: build a request with the _meta every stateless request needs, send it, and read the
reply. Here it calls handle() directly. In 2.6 the same messages go over a real pipe.
CLIENT_META = {"io.modelcontextprotocol/protocolVersion": PROTOCOL_VERSION,
"io.modelcontextprotocol/clientCapabilities": {"elicitation": {"form": {}}},
"io.modelcontextprotocol/clientInfo": {"name": "notes-client", "version": "1.0"}}
next_id = 0
def request(method: str, params: dict | None = None, send=handle) -> dict:
global next_id
next_id += 1
params = {**(params or {}), "_meta": CLIENT_META} if method != "server/discover" else (params or {})
return json.loads(send(json.dumps({"jsonrpc": "2.0", "id": next_id, "method": method, "params": params})))
discover = request("server/discover")["result"]
print("server/discover ->", discover["supportedVersions"], list(discover["capabilities"]))
tools = request("tools/list")["result"]["tools"]
print("tools/list ->", [t["name"] for t in tools])
ok = request("tools/call", {"name": "get_order_status", "arguments": {"order_id": "A100"}})["result"]
print("call A100 ->", ok["structuredContent"], "isError:", ok["isError"])
failed = request("tools/call", {"name": "get_order_status", "arguments": {"order_id": "Z9"}})["result"]
print("call Z9 ->", failed["content"][0]["text"], "isError:", failed["isError"])
bad = request("tools/call", {"name": "get_order_status", "arguments": {"id": 100}})["result"]
print("bad arguments ->", bad["content"][0]["text"])
unknown = request("tools/call", {"name": "delete_everything", "arguments": {}})
print("unknown tool ->", unknown["error"])
policy = request("resources/read", {"uri": "policy://refunds"})["result"]["contents"][0]["text"]
print("resources/read ->", policy)
server/discover -> ['2026-07-28'] ['tools', 'resources']
tools/list -> ['get_order_status', 'cancel_order']
call A100 -> {'order_id': 'A100', 'status': 'shipped', 'eta': 'Friday'} isError: False
call Z9 -> Error: order Z9 not found: IDs look like A100 isError: True
bad arguments -> Error: missing 'order_id'; unexpected 'id'
unknown tool -> {'code': -32602, 'message': 'unknown tool: delete_everything'}
resources/read -> Refunds within 7 days of delivery. Damaged items: full refund.
And the multi round-trip cancel, where the "user" is a function that always says yes:
def ask_user(question: dict) -> dict:
print(" host shows:", question["params"]["message"])
return {"action": "accept", "content": {"confirm": True}}
def call_with_input(name: str, arguments: dict) -> dict:
responses = None
for round_no in range(1, 4): # cap the rounds, like the SDK does
params = {"name": name, "arguments": arguments}
if responses:
params["inputResponses"] = responses
result = request("tools/call", params)["result"]
print(f"round {round_no}: resultType={result['resultType']}")
if result["resultType"] != "input_required":
return result
responses = {key: ask_user(q) for key, q in result["inputRequests"].items()}
raise RuntimeError("too many input rounds")
final = call_with_input("cancel_order", {"order_id": "A101"})
print(final["content"][0]["text"], "| A101 is now", ORDERS["A101"]["status"])
round 1: resultType=input_required
host shows: Really cancel order A101?
round 2: resultType=complete
A101 cancelled | A101 is now cancelled
Finally, a client on an old spec version is refused with a clear error instead of a confusing failure later:
old = json.loads(handle(json.dumps({
"jsonrpc": "2.0", "id": 99, "method": "tools/list",
"params": {"_meta": {"io.modelcontextprotocol/protocolVersion": "2025-06-18"}}})))
print(old["error"])
{'code': -32022, 'message': "unsupported protocol version '2025-06-18'", 'data': {'supported': ['2026-07-28']}}
2.6 The stdio transport, for real¶
For a local server, the host starts it as a child process and they exchange one JSON message per line over stdin and stdout. Anything else the server prints must go to stderr, or it corrupts the stream.
Let's write a minimal server file and talk to it through a real pipe:
import subprocess
import sys
import textwrap
SERVER_FILE = textwrap.dedent('''
import json, sys
def handle(msg):
if msg["method"] == "tools/list":
return {"tools": [{"name": "ping", "description": "Health check",
"inputSchema": {"type": "object", "properties": {}}}]}
if msg["method"] == "tools/call":
return {"content": [{"type": "text", "text": "pong"}], "isError": False}
return None
print("orders server started", file=sys.stderr) # logs go to stderr, never stdout
for line in sys.stdin: # one JSON-RPC message per line
msg = json.loads(line)
result = handle(msg)
reply = ({"jsonrpc": "2.0", "id": msg["id"], "result": {**result, "resultType": "complete"}}
if result is not None else
{"jsonrpc": "2.0", "id": msg["id"], "error": {"code": -32601, "message": "method not found"}})
print(json.dumps(reply), flush=True)
''')
with open("tiny_server.py", "w") as f:
f.write(SERVER_FILE)
proc = subprocess.Popen([sys.executable, "tiny_server.py"], stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
def send_line(raw: str) -> str:
proc.stdin.write(raw + "\n")
proc.stdin.flush()
return proc.stdout.readline()
print(request("tools/list", send=send_line)["result"]["tools"][0]["name"])
print(request("tools/call", {"name": "ping", "arguments": {}}, send=send_line)["result"]["content"][0]["text"])
print(request("prompts/list", send=send_line)["error"]["message"])
proc.stdin.close()
proc.wait()
print("server stderr:", proc.stderr.read().strip())
That's the whole stdio transport. Claude Desktop, Cursor and Claude Code start local servers exactly like this, from a command in their config file (page 4). Remote servers use Streamable HTTP instead: the same JSON, sent as HTTP POSTs (page 5).
Interview questions¶
What's the difference between a JSON-RPC error and isError: true in MCP?
A JSON-RPC error means the request couldn't be processed: unknown method, invalid params, an unknown tool,
an unsupported protocol version. isError: true inside a normal result means the tool ran and failed (order not
found, API down). The model sees tool errors and can correct itself; protocol errors are for the client code.
How does a stateless MCP server ask the user a question in the middle of a tool call?
With the multi round-trip pattern. The server returns resultType: "input_required" with inputRequests (for
example an elicitation form). The client collects the answers and re-sends the same request with
inputResponses. The server keeps no state between rounds, so any replica can handle the retry. Servers can
also return an opaque requestState that the client echoes back.
Why must a stdio MCP server never print to stdout?
Stdout is the protocol channel: the client parses every line as a JSON-RPC message. A stray print() corrupts
the stream and breaks the connection. Logs go to stderr, which hosts usually capture into a log file.
What does server/discover return and when is it used?
The protocol versions the server supports, its capabilities (tools, resources, prompts, extensions), its
identity and optional instructions. Clients can call it first to choose a version, and SDKs use it as a probe:
if a server doesn't understand it, the client falls back to the older initialize handshake.
Practice¶
- Add a
prompts/listandprompts/gethandler with one prompt,apology_email(customer, issue). - Make
ask_userdecline. Check that A100 is kept and the model gets a readable message. - Add an
amount_inrinteger argument to a newrefund_ordertool and extendcheck_argsto validate integers.
Next: 3. Building servers with the Python SDK Back to: MCP overview · Notes overview