標準ライブラリだけでMCP stdioサーバーを作る

目標

json・sys・sqlite3だけで、MCPサーバーを1本書きます。クライアントが送ってくるinitialize → notifications/initialized → tools/list → tools/callを順番に受け付け、最後にそのサーバーを子プロセスとして起動するクライアントまで書きます。

なぜ重要なのか

SDKを使えば5行で終わる作業です。しかし事故は、SDKが隠してくれていた部分で起きます。stdoutに出力したデバッグ用の1行がクライアントのJSONパーサーを壊し、通知に返答を送って対応相手のいないレスポンスが生まれ、ツールが失敗したのにプロトコルエラーで返したためにモデルが「サーバーが壊れた」と判断してしまいます。フレーミング(1行 = 1メッセージ)、idによる対応付け、通知の無視、2種類のエラーの区別。この4つがMCPの実体であり、手で一度組んでみると、SDKのログが何を言っているのか読めるようになります。

ステップ

シードSQL(/root/mcp/seed.sql)を保存し、DB(/root/mcp/shop.db)へ読み込んでください。customersが5行、ordersが8行になっている必要があります。
サーバーのファイル(/root/mcp/server.py)を作成してください。stdinから1行ずつJSON-RPCリクエストを読み、initializeには、protocolVersionが2025-06-18で、capabilities.toolsとserverInfo.nameを持つresultで返答する必要があります。
同じファイルで、通知(idのないメッセージ)には返答せず、未知のメソッドには-32601、壊れたJSONには-32700のエラー(idはnull)で返答するようにしてください。サーバーが落ちてはいけません。
tools/listにツールを2つ返してください。list_customersとcount_orders(inputSchema.properties.status、文字列、required)です。各ツールに、descriptionと、type: objectのinputSchemaが必要です。
tools/callを実装してください。count_ordersに{"status":"paid"}を渡すと、DBのpaidの件数がcontent[0].textに入っている必要があり、list_customersは5人の顧客名をすべて含める必要があります。DBのパスは、環境変数MCP_DBがあればそれを、なければ/root/mcp/shop.dbを使います。
エラーを2種類に分けてください。存在しないツール名にはJSON-RPCエラー-32602を返し、count_ordersに未知の状態値(例: banana)を渡したときは、resultにisError: trueと説明のテキストを入れて返す必要があります。
クライアントのファイル(/root/mcp/client.py)を作成してください。サーバーをsubprocessで起動し、initialize → notifications/initialized → tools/list → tools/call(count_orders、paid)を送って、結果を、tools(名前の一覧)とpaid_orders(レスポンスのテキスト)として、ファイル(/root/mcp/session.json)に書き込みます。環境変数MCP_SESSION_OUTがあれば、そのパスに書きます。
サーバーのすべてのログをstderrに送ってください。リクエストごとにメソッド名がstderrに1行ずつ出力され、stdoutにはJSONのレスポンス以外に何も出てはいけません。

参考

1行ずつ読むループはfor line in sys.stdin:で足ります。レスポンスはsys.stdout.write(json.dumps(...) + "\n")のあとに必ずflush()してください。パイプは行バッファリングではないため、flushがないとクライアントは永遠に待ちます。
通知かどうかは"id" in msgで判断します。msg.get("id")がNoneかどうかで見ると、idが実際にnullである(不正な)リクエストと区別がつきません。
json.loadsが投げる例外を捕まえないと、壊れた行が1つあるだけでサーバーが丸ごと落ちます。クライアントはそれを「サーバーが消えた」と受け取ります。
手で試すには、次のようにします。printf '%s\n' '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{}}' | python3 /root/mcp/server.py
よくある間違い1: print()でデバッグ用の文をstdoutに出力してしまうことです。ステップ8の採点がそれを検出します。
よくある間違い2: 存在しないツールをisError: trueで返してしまうことです。それはツールが実行されている途中で失敗したのではなく、リクエストそのものが間違っているので、プロトコルエラーです。

店のDBを作る

シードSQL(/root/mcp/seed.sql)を保存し、DB(/root/mcp/shop.db)へ読み込んでください。customersが5行、ordersが8行になっている必要があります。

sqlite3では、sqlite3 shop.db < seed.sqlでファイルをまるごと実行します。Pythonで行うなら、sqlite3.connect(...).executescript(open(...).read())です。すでにあるDBに再度読み込むと、テーブルが存在するというエラーになるので、先に削除してください。

initializeに返答する

サーバーのファイル(/root/mcp/server.py)を作成してください。stdinから1行ずつJSON-RPCリクエストを読み、initializeには、protocolVersionが2025-06-18で、capabilities.toolsとserverInfo.nameを持つresultで返答する必要があります。

骨組みの例のループで、method == "initialize"のときにresultの辞書を作り、reply(msg["id"], result=...)で送ります。resultには、protocolVersion、capabilities(toolsキーを持つオブジェクト)、serverInfo(name・version)の3つが入ります。

通知には返答しない

同じファイルで、通知(idのないメッセージ)には返答せず、未知のメソッドには-32601、壊れたJSONには-32700のエラー(idはnull)で返答するようにしてください。サーバーが落ちてはいけません。

JSON-RPCでは、通知はidメンバーのないリクエストであり、サーバーは返答してはいけません。json.loadsをtryで囲み、JSONDecodeErrorならidをnullにして-32700を送り、continueしてください。それ以外の例外も捕まえて-32603で返答すれば、サーバーは生き残ります。

ツール一覧を返す

tools/listにツールを2つ返してください。list_customersとcount_orders(inputSchema.properties.status、文字列、required)です。各ツールに、descriptionと、type: objectのinputSchemaが必要です。

resultは{"tools": [...]}で、ツール1つはname・description・inputSchemaの3つのキーを持ちます。inputSchemaはJSON Schemaのオブジェクトなので、"type": "object"とpropertiesを持ちます。引数のないツールにも、properties: {}は置きます。

ツールを実際に実行する

tools/callを実装してください。count_ordersに{"status":"paid"}を渡すと、DBのpaidの件数がcontent[0].textに入っている必要があり、list_customersは5人の顧客名をすべて含める必要があります。DBのパスは、環境変数MCP_DBがあればそれを、なければ/root/mcp/shop.dbを使います。

paramsは{"name": 도구이름, "arguments": {...}}の形です(プレースホルダーはツール名です)。結果は{"content": [{"type": "text", "text": "..."}]}の形にしてください。os.environ.get("MCP_DB", "/root/mcp/shop.db")でパスを決め、リクエストごとにsqlite3.connectして閉じてください。

エラーを2種類に分ける

エラーを2種類に分けてください。存在しないツール名にはJSON-RPCエラー-32602を返し、count_ordersに未知の状態値(例: banana)を渡したときは、resultにisError: trueと説明のテキストを入れて返す必要があります。

仕様は、「未知のツール・不正な引数」をプロトコルエラー(errorメンバー、例のコードは-32602)、「ツールが実行されている途中で失敗したもの」をツール実行エラー(result内のisError: true)として区別しています。後者は、モデルが読んで再試行できるように、テキストで理由を書きます。

クライアントで1つのセッションを通す

クライアントのファイル(/root/mcp/client.py)を作成してください。サーバーをsubprocessで起動し、initialize → notifications/initialized → tools/list → tools/call(count_orders、paid)を送って、結果を、tools(名前の一覧)とpaid_orders(レスポンスのテキスト)として、ファイル(/root/mcp/session.json)に書き込みます。環境変数MCP_SESSION_OUTがあれば、そのパスに書きます。

subprocess.Popen([...], stdin=PIPE, stdout=PIPE, text=True)で起動し、送るたびにstdin.flush()し、受け取るときはstdout.readline()で1行をjson.loadsします。通知を送ったあとはreadlineしないでください。返答が来ないので永遠に待つことになります。終了するときは、stdinを閉じてwait()します。

stdoutをプロトコル専用にする

サーバーのすべてのログをstderrに送ってください。リクエストごとにメソッド名がstderrに1行ずつ出力され、stdoutにはJSONのレスポンス以外に何も出てはいけません。

sys.stderr.write(...)またはprint(..., file=sys.stderr)を使います。仕様(stdioトランスポート)は、サーバーがstdoutに有効なMCPメッセージ以外を書いてはならず、stderrはログ用途に使ってよいと定めています。採点ツールは、stdoutのすべての行をJSONとしてパースしてみます。