Implementing JSON-RPC over stdio Protocol in MCP Station
MCP (Model Context Protocol) uses JSON-RPC 2.0 over stdio as its communication protocol. This post summarizes what I learned while implementing this protocol in MCP Station.
JSON-RPC 2.0 Basic Structure
JSON-RPC has a simple but well-defined spec. The shapes of requests and responses are fixed.
from typing import Any, Optional
from dataclasses import dataclass
@dataclass
class JsonRpcRequest:
method: str
params: dict = None
id: Optional[int] = None
jsonrpc: str = "2.0"
def to_dict(self) -> dict:
d = {"jsonrpc": self.jsonrpc, "method": self.method}
if self.params is not None:
d["params"] = self.params
if self.id is not None:
d["id"] = self.id
return d
@dataclass
class JsonRpcResponse:
id: Optional[int]
result: Any = None
error: Optional[dict] = None
jsonrpc: str = "2.0"
@property
def is_error(self) -> bool:
return self.error is not None
stdio Communication Handler
Communication with the MCP server goes through stdin/stdout, exchanging one JSON object per line.
import asyncio
import json
from asyncio import StreamReader, StreamWriter
class StdioTransport:
def __init__(self, process: asyncio.subprocess.Process):
self.process = process
self._request_id = 0
self._pending: Dict[int, asyncio.Future] = {}
def _next_id(self) -> int:
self._request_id += 1
return self._request_id
async def send_request(self, method: str, params: dict = None) -> Any:
request_id = self._next_id()
request = JsonRpcRequest(
method=method, params=params, id=request_id
)
future = asyncio.get_event_loop().create_future()
self._pending[request_id] = future
line = json.dumps(request.to_dict()) + "\n"
self.process.stdin.write(line.encode())
await self.process.stdin.drain()
return await asyncio.wait_for(future, timeout=30.0)
async def _read_loop(self):
while True:
line = await self.process.stdout.readline()
if not line:
break
try:
data = json.loads(line.decode())
if "id" in data and data["id"] in self._pending:
future = self._pending.pop(data["id"])
if "error" in data:
future.set_exception(
JsonRpcError(data["error"])
)
else:
future.set_result(data.get("result"))
elif "method" in data:
# Handle notifications sent from the server
await self._handle_notification(data)
except json.JSONDecodeError:
continue
MCP Initialization Handshake
The first step in the MCP protocol is the initialize request, where the server and client exchange their respective capabilities.
class MCPClient:
def __init__(self, transport: StdioTransport):
self.transport = transport
self.server_capabilities = {}
async def initialize(self) -> dict:
result = await self.transport.send_request("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {
"roots": {"listChanged": True},
},
"clientInfo": {
"name": "mcp-station",
"version": "1.0.0",
},
})
self.server_capabilities = result.get("capabilities", {})
# Notify that initialization is complete
await self.transport.send_request("notifications/initialized")
return result
async def list_tools(self) -> list:
result = await self.transport.send_request("tools/list")
return result.get("tools", [])
async def call_tool(self, name: str, arguments: dict) -> Any:
result = await self.transport.send_request("tools/call", {
"name": name,
"arguments": arguments,
})
return result
Error Handling Strategy
stdio communication is more reliable than network communication, but it still has failure modes like process crashes and broken pipes.
class JsonRpcError(Exception):
PARSE_ERROR = -32700
INVALID_REQUEST = -32600
METHOD_NOT_FOUND = -32601
INTERNAL_ERROR = -32603
def __init__(self, error_data: dict):
self.code = error_data.get("code", self.INTERNAL_ERROR)
self.message = error_data.get("message", "Unknown error")
super().__init__(f"[{self.code}] {self.message}")
Wrapping Up
JSON-RPC over stdio is straightforward to implement, and equally straightforward to debug — logging stdin/stdout gives you full visibility into every message exchanged. As the MCP ecosystem grows, a solid understanding of this protocol is becoming increasingly valuable.