前兩天講了架構和原語。今天完全不用 MCP SDK，只用 Python 標準函式庫，手寫一個 MCP Server 和一個 Client。
然後拿那支手刻 client 去打 Day 10 用官方 SDK 寫的 server。能對話，就證明 SDK 只是包裝。
MCP 建立在 JSON-RPC 2.0 上。Request 帶 id 和 method；Response 必須帶回同一個 id，而且 result 和 error 只能有一個。
第三種是 Notification，跟 Request 唯一的差別是沒有 id，意思是收到的一方絕對不可以回應。有 id 才要回，沒 id 回了就是 bug。
JSON-RPC 也是對稱的，Server 可以主動發 request 給 Client，昨天講的 Sampling、Elicitation 就靠這點運作。
MCP 沒有 LSP 那種 Content-Length 標頭，stdio 上一行一則訊息，換行結束：
def send(msg: dict[str, Any]) -> None:
sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n")
sys.stdout.flush()
初始化時 Client 送 initialize，Server 回協定版本和 capabilities，Client 再送一則沒有 id 的 notifications/initialized，握手才算完成。
if method == "notifications/initialized":
initialized = True
log("握手完成，客戶端說它準備好了（這是 notification，我不回應）")
continue
if not initialized and method != "ping":
err(req_id, -32002, "尚未完成 initialize 握手", {"method": method})
continue
手刻 server 在收到 notifications/initialized 之前，除了 initialize 和 ping，一律用自訂錯誤碼 -32002 拒絕。
關閉階段沒有 shutdown 方法，這跟 LSP 不一樣，MCP 把關閉交給傳輸層，stdio 就是 client 關掉 stdin。
raw_client.py 會把送出（→）和收回（←）的訊息都印出來。先打手刻的 server，再加上 --sdk 打 Day 10 的 devbench。
實測輸出：
對象：同資料夾裡手刻的 raw_server (raw_server.py)
→ {"jsonrpc": "2.0", "method": "notifications/initialized"}
沒有 ← 那一行，因為 notification 本來就不該有回應。
(a) 協定層錯誤：方法不存在 → JSON-RPC error 物件
→ {"jsonrpc": "2.0", "id": 4, "method": "tools/nonexistent"}
← {"jsonrpc": "2.0", "id": 4, "error": {"code": -32601, "message": "Method not found", "data": {"method": "tools/nonexistent"}}}
(b) 應用層錯誤：工具本身執行失敗 → 正常的 result，但 isError=true
→ {"jsonrpc": "2.0", "id": 5, "method": "tools/call", "params": {"name": "nope", "arguments": {}}}
← {"jsonrpc": "2.0", "id": 5, "result": {"content": [{"type": "text", "text": "沒有這個工具：nope"}], "isError": true}}
對象：Day 10 用官方 SDK 寫的 devbench server (server.py)
對方是 devbench v0.1.0
協定版本 2025-06-18
→ {"jsonrpc": "2.0", "id": 2, "method": "tools/list"}
← {"jsonrpc":"2.0","id":2,"result":{"tools":[{"annotations":{"destructiveHint":false,"idempotentHint":true,"readOnlyHint":true},"description":"讀取專案內的一個檔案並回傳內容（附行號）。","inputSchema":{"
兩種錯誤的處理方式完全相反。協定層錯誤只有 error，代表請求本身有問題，要給開發者修程式。應用層錯誤有 result，代表請求沒問題但執行失敗，應該原封不動餵回給模型。
如果當成協定錯誤丟出去，agent 迴圈直接中斷，模型連自我修正的機會都沒有。Day 13 實作 ReAct 時會再用到。
換成官方 SDK 寫的 server，同一支 client 握手、列工具、呼叫工具、兩種錯誤全部照走。回應裡的 "readOnlyHint":true 是 camelCase，Day 10 會看到同一個欄位在 Python 物件上叫 read_only_hint。
日誌不准印到 stdout。stdout 是協定通道，隨手 print 一行 debug，對方就會把它當成 JSON-RPC 訊息解析。實測 SDK 2.2.0 的 client 會噴一段解析錯誤、跳過那行，連線沒斷；今天這支手刻 client 直接 json.loads，則會當場拋例外。所有日誌一律走 stderr。
記得 flush。Python 的 stdout 接到 pipe 時是整塊緩衝的，不 flush 的話小訊息會一直卡在緩衝區，看起來就像對方沒回應。
明天講這些訊息怎麼送到對方手上：傳輸協定 stdio vs Streamable HTTP，有狀態與無狀態。順便看看為什麼 server 設成 stateless，不代表你的程式就沒有狀態。
days/day08_jsonrpc/raw_server.py"""Day 8：一個完全不用 SDK 的 MCP Server。
只有標準函式庫。目的是讓你看清楚：所謂 MCP Server，就是一個
「從 stdin 讀一行 JSON、往 stdout 寫一行 JSON」的迴圈。
MCP 的 stdio 傳輸用的是 newline-delimited JSON（一行一則訊息），
不是 LSP 那種 Content-Length 標頭。日誌一律寫 stderr，
因為 stdout 是協定專用通道，印上去的每一行都會被對方當成訊息解析。
"""
from __future__ import annotations
import json
import sys
from typing import Any
PROTOCOL_VERSION = "2025-06-18"
TOOLS = [
{
"name": "add",
"description": "把兩個整數相加",
"inputSchema": {
"type": "object",
"properties": {
"a": {"type": "integer", "description": "第一個加數"},
"b": {"type": "integer", "description": "第二個加數"},
},
"required": ["a", "b"],
},
},
{
"name": "echo",
"description": "原封不動回傳輸入的文字",
"inputSchema": {
"type": "object",
"properties": {"text": {"type": "string"}},
"required": ["text"],
},
},
]
def log(msg: str) -> None:
print(f"[raw_server] {msg}", file=sys.stderr, flush=True)
def send(msg: dict[str, Any]) -> None:
sys.stdout.write(json.dumps(msg, ensure_ascii=False) + "\n")
sys.stdout.flush()
def ok(req_id: Any, result: dict[str, Any]) -> None:
send({"jsonrpc": "2.0", "id": req_id, "result": result})
def err(req_id: Any, code: int, message: str, data: Any = None) -> None:
error: dict[str, Any] = {"code": code, "message": message}
if data is not None:
error["data"] = data
send({"jsonrpc": "2.0", "id": req_id, "error": error})
def call_tool(name: str, args: dict[str, Any]) -> dict[str, Any]:
if name == "add":
total = int(args["a"]) + int(args["b"])
return {"content": [{"type": "text", "text": str(total)}], "isError": False}
if name == "echo":
return {"content": [{"type": "text", "text": str(args["text"])}], "isError": False}
# 工具「執行失敗」不是協定錯誤，要用 isError 回報，不是回 JSON-RPC error
return {"content": [{"type": "text", "text": f"沒有這個工具：{name}"}], "isError": True}
def main() -> None:
log("等待 stdin…")
initialized = False
for line in sys.stdin:
line = line.strip()
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError as exc:
err(None, -32700, "Parse error", str(exc))
continue
method = msg.get("method")
req_id = msg.get("id") # 沒有 id = notification，不可以回應
params = msg.get("params") or {}
log(f"收到 {method}（id={req_id}）")
# ── 生命週期 ─────────────────────────────────────────────
if method == "initialize":
ok(req_id, {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {"tools": {"listChanged": False}},
"serverInfo": {"name": "raw-server", "version": "0.0.1"},
"instructions": "這是一個不用 SDK 手刻的示範 server。",
})
continue
if method == "notifications/initialized":
initialized = True
log("握手完成，客戶端說它準備好了（這是 notification，我不回應）")
continue
# ── 握手前不准做事 ───────────────────────────────────────
if not initialized and method != "ping":
err(req_id, -32002, "尚未完成 initialize 握手", {"method": method})
continue
if method == "ping":
ok(req_id, {})
elif method == "tools/list":
ok(req_id, {"tools": TOOLS})
elif method == "tools/call":
try:
ok(req_id, call_tool(params["name"], params.get("arguments", {})))
except KeyError as exc:
err(req_id, -32602, "Invalid params", f"缺少欄位 {exc}")
else:
err(req_id, -32601, "Method not found", {"method": method})
log("stdin 關閉，結束。")
if __name__ == "__main__":
main()
days/day08_jsonrpc/raw_client.py"""Day 8：手寫 JSON-RPC client，把 MCP 握手的每一個 byte 攤開來看。
這支 client 有兩個模式：
uv run python days/day08_jsonrpc/raw_client.py # 打自己人（raw_server.py）
uv run python days/day08_jsonrpc/raw_client.py --sdk # 打 Day 10 的 SDK server
第二個模式是重點：同一支手刻 client，可以直接跟官方 SDK 寫的 server
對話。這證明 SDK 沒有另外發明什麼，它就是照 JSON-RPC 2.0 在講話。
"""
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
from typing import Any
HERE = Path(__file__).resolve()
ROOT = HERE.parents[2]
RAW_SERVER = HERE.with_name("raw_server.py")
SDK_SERVER = ROOT / "days" / "day10_mcp_server" / "server.py"
PROTOCOL_VERSION = "2025-06-18"
class RawSession:
"""一個 JSON-RPC over stdio 的最小客戶端。"""
def __init__(self, server: Path) -> None:
self.proc = subprocess.Popen(
[sys.executable, str(server)],
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, # server 的日誌走 stderr，這裡先丟掉
text=True,
bufsize=1,
cwd=str(ROOT),
)
self._next_id = 0
# ── 低階：送一行、收一行 ────────────────────────────────────
def _write(self, msg: dict[str, Any]) -> None:
line = json.dumps(msg, ensure_ascii=False)
print(f" → {line[:180]}")
assert self.proc.stdin
self.proc.stdin.write(line + "\n")
self.proc.stdin.flush()
def _read(self) -> dict[str, Any]:
assert self.proc.stdout
line = self.proc.stdout.readline()
if not line:
raise RuntimeError("server 關閉了連線")
print(f" ← {line.strip()[:180]}")
return json.loads(line)
# ── 高階：request 要 id，notification 不要 ──────────────────
def request(self, method: str, params: dict[str, Any] | None = None) -> dict[str, Any]:
self._next_id += 1
self._write({
"jsonrpc": "2.0",
"id": self._next_id,
"method": method,
**({"params": params} if params is not None else {}),
})
resp = self._read()
assert resp["id"] == self._next_id, "id 必須配對，否則你就串線了"
return resp
def notify(self, method: str, params: dict[str, Any] | None = None) -> None:
"""notification 沒有 id，對方不會、也不准回應。"""
self._write({
"jsonrpc": "2.0",
"method": method,
**({"params": params} if params is not None else {}),
})
def close(self) -> None:
if self.proc.stdin:
self.proc.stdin.close()
self.proc.wait(timeout=10)
def rule(title: str) -> None:
print()
print("─" * 72)
print(title)
print("─" * 72)
def walk_lifecycle(server: Path, label: str) -> None:
print("=" * 72)
print(f"對象：{label} ({server.name})")
print("=" * 72)
s = RawSession(server)
# ── ① initialize ────────────────────────────────────────────
rule("① initialize：交換協定版本與能力")
resp = s.request("initialize", {
"protocolVersion": PROTOCOL_VERSION,
"capabilities": {},
"clientInfo": {"name": "raw-client", "version": "0.0.1"},
})
result = resp["result"]
info = result.get("serverInfo") or result.get("server_info") or {}
print(f"\n 對方是 {info.get('name')} v{info.get('version')}")
print(f" 協定版本 {result.get('protocolVersion') or result.get('protocol_version')}")
print(f" 宣告的能力 {list(result.get('capabilities', {}))}")
# ── ② initialized（notification）─────────────────────────────
rule("② notifications/initialized：沒有 id，所以沒有回應")
s.notify("notifications/initialized")
print("\n 沒有 ← 那一行，因為 notification 本來就不該有回應。")
print(" 這是 JSON-RPC 最容易搞錯的地方：有 id 才要回，沒 id 回了就是 bug。")
# ── ③ tools/list ───────────────────────────────────────────
rule("③ tools/list：問對方有什麼工具")
resp = s.request("tools/list")
tools = resp["result"]["tools"]
print(f"\n 拿到 {len(tools)} 個工具：")
for t in tools:
schema = t.get("inputSchema") or t.get("input_schema") or {}
print(f" {t['name']:14s} {(t.get('description') or '').splitlines()[0][:40]}")
print(f" {'':14s} 必填參數 {schema.get('required', [])}")
# ── ④ tools/call ───────────────────────────────────────────
rule("④ tools/call：真的叫一次")
name, args = ("add", {"a": 17, "b": 25}) if server == RAW_SERVER else ("list_files", {"pattern": "src/**/*.py", "limit": 3})
resp = s.request("tools/call", {"name": name, "arguments": args})
content = resp["result"].get("content", [])
print(f"\n {name}({args}) 回傳：")
for block in content:
for line in (block.get("text") or "").splitlines()[:6]:
print(f" {line}")
# ── ⑤ 錯誤 ─────────────────────────────────────────────────
rule("⑤ 兩種完全不同的「錯誤」")
print("\n (a) 協定層錯誤：方法不存在 → JSON-RPC error 物件")
resp = s.request("tools/nonexistent")
e = resp.get("error", {})
print(f" code={e.get('code')} message={e.get('message')}")
print(" -32601 是 JSON-RPC 標準碼：Method not found")
print("\n (b) 應用層錯誤：工具本身執行失敗 → 正常的 result，但 isError=true")
bad_name, bad_args = ("nope", {}) if server == RAW_SERVER else ("read_file", {"path": "../../../etc/passwd"})
resp = s.request("tools/call", {"name": bad_name, "arguments": bad_args})
r = resp.get("result", {})
print(f" 有 result 嗎？{'result' in resp} isError={r.get('isError') or r.get('is_error')}")
print(f" 內容 → {(r.get('content') or [{}])[0].get('text', '')[:80]}")
print("\n ★ 分清楚這兩種，你的 agent 才知道該重試、還是該告訴使用者。")
s.close()
print()
def main() -> None:
if "--sdk" in sys.argv:
walk_lifecycle(SDK_SERVER, "Day 10 用官方 SDK 寫的 devbench server")
print("=" * 72)
print("★ 同一支手刻 client，打得動官方 SDK 寫的 server。")
print(" 這就是「協定」的意義：兩邊只要遵守同一份規格，就能互通。")
print("=" * 72)
else:
walk_lifecycle(RAW_SERVER, "同資料夾裡手刻的 raw_server")
print("=" * 72)
print("接著試試看：uv run python days/day08_jsonrpc/raw_client.py --sdk")
print("同一支 client 去打 Day 10 用 SDK 寫的 server，會一模一樣地動。")
print("=" * 72)
if __name__ == "__main__":
main()
iThome鐵人賽
看影片追技術