2026年10月上旬の話です。私はMCPサーバー経由で日誌DBを更新する夜間エージェントを運用していました。23時の起動は正常、ところが午前2時57分にサーバープロセスが静かに死んでエージェントが沈黙しました。ログはBrokenPipeErrorの1行だけで、原因が分からないまま2日を溶かし、朝までに4本のジョブが全滅しました。原因は「MCPサーバーは死なない」という前提で組んだstdio接続でした。この記事では、死因を残す方法と、指数バックオフで自動再接続するクライアントの完全実装（標準ライブラリだけ・コピペで動く）を載せます。
何が起きていたか: 前半は走るのに後半で全滅する
まず実測のタイムラインを表にまとめました。
| 時刻 | 状態 | ログに残ったもの |
|---|---|---|
| 23:04 | 初回起動、initialize成功 | 起動ログのみ |
| 02:57 | サーバープロセス死亡（rc=-9） | 死因はstderr側。しかし捨てていたので不明 |
| 02:58 | ジョブ1がBrokenPipeErrorで終了 | traceback 1行、再試行なし |
| 03:40 | ジョブ2〜4も同じ箇所で沈黙 | リカバリは朝の手動作業になった |
MCPのstdioトランスポートは「クライアントがサーバーを子プロセスとして立て、標準入出力でJSON-RPCを流す」構成です。子プロセスは自分より長く生きる保証がどこにもなく、メモリ枯渇でのkillやライブラリの未処理例外で夜間に死にます。死ぬこと自体は防げなくても、検知と復旧はこちらで設計する必要がありました。
失敗した設計: 再接続なし・stderr破棄のstdioクライアント
最初に書いた駄目なコードがこちらです。これは失敗パターンなのでそのままコピペしないでください（正しい形は次章に完全形があります）。
# ❌ 失敗設計: 再接続なし・stderr破棄・死因調査不能のstdioクライアント
import json
import subprocess
proc = subprocess.Popen(["python3", "mcp_server.py"],
stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, # ← 死因がここを流れるのに捨てている
text=True, encoding="utf-8", bufsize=1)
def call_tool(name, arguments):
proc.stdin.write(json.dumps({"jsonrpc": "2.0", "id": 1,
"method": "tools/call",
"params": {"name": name, "arguments": arguments}}) + "\n")
proc.stdin.flush()
resp = json.loads(proc.stdout.readline()) # ← プロセス死亡でEOFになり、ここで落ちる
return resp.get("result")
print(call_tool("search", {"query": "日誌"})) # 数時間は順調に動く
print(call_tool("search", {"query": "月次"})) # ← サーバー死亡後、ここでBrokenPipeError
このコードの決定的な失敗は次の3点です。
- 再接続の発想が無い — 子プロセスは1度立てたら永久に生きる前提。死んだら例外が出るだけで、起動し直して同じ要求をやり直す層が無いためジョブごと落ちる
- stderrをDEVNULLで破棄 — 事故の死因（out of memory・未処理例外など）はstderrに流れる。捨てた結果、2日間の原因究明は完全に暗号解読になった
-
stdoutのEOFを無加工でreadline — プロセス死亡後は空文字を返すので、
json.loadsが落ちて本命のエラーメッセージすら残らない
直し方1: 再接続つきstdioクライアント（標準ライブラリだけの完全形）
要求のやり直し・プロセス再起動・initializeのやり直し・stderrの保存・指数バックオフを、1ファイルに全部入れました。末尾の__main__には「2回目のtools/callで自爆するデモサーバー」まで同梱してあるので、そのまま実行すれば再接続の動作を観察できます。
#!/usr/bin/env python3
# ✅ 完全形: 再接続つきMCP stdioクライアント（標準ライブラリのみ・コピペで動く）
import collections
import json
import queue
import random
import subprocess
import threading
import time
class McpStdioClient:
"""stdioトランスポートのMCPクライアント。
サーバープロセスの死亡と書き込み断線を検知したら、指数バックオフで
再起動 → initialize → 同じ要求のやり直しまで自動で行う。"""
def __init__(self, command, max_retries=5, base_delay=1.0):
self.command = list(command) # 例: ["python3", "/tmp/mcp_demo_flaky_server.py"]
self.max_retries = max_retries
self.base_delay = base_delay
self.proc = None
self.reads = queue.Queue() # 現世代のサーバー → クライアント行
self.err_buf = collections.deque(maxlen=20) # 死因調査用のstderr末尾
self.threads = []
self.next_id = 0
def start(self):
"""プロセス起動＋initializeまで。死んだプロセスからの復帰でも同じ関数を使う。"""
self.proc = subprocess.Popen(
self.command,
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, encoding="utf-8", bufsize=1,
)
self.reads = queue.Queue() # 世代が変わるので積み直す
self.err_buf.clear()
self.threads = [
threading.Thread(target=self._pump, args=(self.proc.stdout, self.reads), daemon=True),
threading.Thread(target=self._pump, args=(self.proc.stderr, self.err_buf), daemon=True),
]
for th in self.threads:
th.start()
self.call("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "retry-client", "version": "1.0"},
}, timeout=15)
def _pump(self, pipe, sink):
"""パイプを1行ずつ拾ってsinkへ。プロセス死亡時はEOF/例外で自然に終わる。"""
try:
for line in pipe:
text = line.strip()
if not text:
continue
# 出口がQueueなら応答流、dequeならstderr保存と使い分ける
if isinstance(sink, queue.Queue):
sink.put(text)
else:
sink.append(text)
except (ValueError, OSError):
pass
if isinstance(sink, queue.Queue):
sink.put(None) # 終了合図
def _send(self, obj):
if self.proc is None or self.proc.poll() is not None:
raise ConnectionError(f"サーバープロセス停止 rc={self._exit_code()}")
try:
self.proc.stdin.write(json.dumps(obj, ensure_ascii=False) + "\n")
self.proc.stdin.flush()
except (BrokenPipeError, OSError) as e:
raise ConnectionError(f"書き込み失敗: {e}") from e
def _exit_code(self):
code = self.proc.returncode if self.proc else None
return code if code is not None else "生存中"
def _wait_response(self, want_id, timeout):
deadline = time.monotonic() + timeout
while True:
remain = deadline - time.monotonic()
if remain <= 0:
raise TimeoutError(f"{timeout}秒以内にid={want_id}の応答なし")
try:
line = self.reads.get(timeout=remain)
except queue.Empty:
continue
if line is None:
raise ConnectionError("サーバーの標準出力が閉じました")
if not line:
continue
try:
msg = json.loads(line)
except json.JSONDecodeError:
continue # 雑音行は無視する
if msg.get("id") != want_id:
continue # 通知とサーバー起点の要求はここでは扱わない
if "error" in msg:
raise RuntimeError(f"MCPエラー: {msg['error']}")
return msg.get("result", {})
def call(self, method, params, timeout=60):
self.next_id += 1
req_id = self.next_id
self._send({"jsonrpc": "2.0", "id": req_id, "method": method, "params": params})
return self._wait_response(req_id, timeout)
def call_tool(self, name, arguments, timeout=60):
result = self.call("tools/call", {"name": name, "arguments": arguments}, timeout=timeout)
return "".join(c.get("text", "") for c in result.get("content", []))
def call_tool_safe(self, name, arguments, timeout=60):
"""本丸。切断とタイムアウトを指数バックオフで自動復旧する。
MCPエラー応答（RuntimeError）は再試行しない: 何度やっても同じ失敗だから。"""
last_error = None
for attempt in range(1, self.max_retries + 1):
try:
if self.proc is None or self.proc.poll() is not None:
self.start() # 未接続・切断済みなら初期化から
return self.call_tool(name, arguments, timeout=timeout)
except (ConnectionError, TimeoutError) as e:
last_error = e
if self.err_buf:
print(f"[retry-client] stderr末尾: {list(self.err_buf)[-3:]}")
delay = self.base_delay * (2 ** (attempt - 1)) + random.uniform(0, 0.5)
print(f"[retry-client] 試行{attempt}失敗: {e} → {delay:.1f}秒待って再接続")
self.shutdown()
time.sleep(delay)
raise ConnectionError(f"{self.max_retries}回の再接続でも通じません: {last_error}")
def shutdown(self):
proc, self.proc = self.proc, None
if proc is None:
return
try:
proc.terminate()
proc.wait(timeout=5)
except Exception:
pass
try:
proc.kill()
except Exception:
pass
# デモ用: 2回目のtools/callで自爆するMCPサーバー（夜間の突然死を再現する）
SERVER_SRC = '''import json
import sys
import os
STATE = {"count": 0}
for line in sys.stdin:
line = line.strip()
if not line:
continue
msg = json.loads(line)
if "id" not in msg:
continue
method = msg.get("method", "")
if method == "initialize":
result = {"protocolVersion": "2024-11-05",
"capabilities": {"tools": {}},
"serverInfo": {"name": "echo", "version": "1.0"}}
elif method == "tools/call":
STATE["count"] += 1
if STATE["count"] >= 2: # 同一プロセスの2回目で自爆（夜間の突然死の再現）
print("FATAL: デモ用に自爆します", file=sys.stderr, flush=True)
os._exit(137)
args_text = json.dumps(msg["params"].get("arguments", {}), ensure_ascii=False)
result = {"content": [{"type": "text", "text": "echo: " + args_text}]}
else:
result = {}
sys.stdout.write(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": result},
ensure_ascii=False) + "\\n")
sys.stdout.flush()
'''
if __name__ == "__main__":
import os
import sys
import tempfile
server_path = os.path.join(tempfile.gettempdir(), "mcp_demo_flaky_server.py")
with open(server_path, "w", encoding="utf-8") as f:
f.write(SERVER_SRC)
client = McpStdioClient([sys.executable, server_path], max_retries=4, base_delay=0.5)
print("1回目:", client.call_tool_safe("echo", {"message": "初回"}))
print("2回目:", client.call_tool_safe("echo", {"message": "自爆後の自動再接続"}))
print("3回目:", client.call_tool_safe("echo", {"message": "再接続後も平気"}))
client.shutdown()
実行すると、2回目と3回目の要求でFATALをstderrから拾い、バックオフして再接続し、同じ要求を失敗ゼロでやり直す様子がそのまま見えます。ポイントは次の3つです。
- stderrは捨てない — dequeに末尾20行だけ残し、異常時に原因の第一手がかりにする
-
MCPエラー応答は再試行しない —
RuntimeErrorはリトライ対象外。切断とタイムアウトだけを復旧対象にする（何度やっても同じ失敗を待つのを避ける） - 世代ごとのqueue — 再起動のたびに応答キューを積み直し、前のプロセスの残骸が後の応答に混入しないようにする
直し方2: 長時間ジョブはStreamable HTTPに寄せる
そもそも論として、夜間バッチのように「子プロセスと数時間共存する」用途はstdioと相性が悪いです。MCPのStreamable HTTPトランスポートなら、サーバーは常駐し、クライアントは1リクエストごとに接続するので、寿命の共有が消えます。残る失敗は「セッションIDの喪失」だけなので、喪失時に初期化からやり直す形にすれば回復します。標準ライブラリだけで書いたのがこちらです。
#!/usr/bin/env python3
# ✅ 完全形: Streamable HTTPの夜間バッチ（セッション喪失を自動回復・標準ライブラリのみ）
import json
import random
import time
import urllib.error
import urllib.request
MCP_URL = "http://localhost:8080/mcp" # FastMCP(mcp run)のデフォルト等のエンドポイント
session_id = None # サーバーから払い出されたセッションID
def _headers():
h = {
"Content-Type": "application/json",
"Accept": "application/json, text/event-stream",
}
if session_id:
h["Mcp-Session-Id"] = session_id
return h
def _post_json(payload, timeout):
global session_id
req = urllib.request.Request(MCP_URL, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers=_headers(), method="POST")
with urllib.request.urlopen(req, timeout=timeout) as resp:
sid = resp.headers.get("Mcp-Session-Id")
if sid:
session_id = sid # 払い出されたら以後の要求に付ける
return resp.read().decode("utf-8", errors="replace")
def _extract_result(payload_text, want_id):
"""応答は素JSONかSSE(data:行)。id一致のレスポンスだけを抜き出す。"""
for line in payload_text.splitlines():
line = line.strip()
if line.startswith("data:"):
line = line[5:].strip()
if not line or not line.startswith("{"):
continue
msg = json.loads(line)
if msg.get("id") == want_id and "result" in msg:
return msg["result"]
raise TimeoutError("応答の中に一致するidがありませんでした")
def notify(method, params):
"""応答を待たない通知。initializedの送付などに使う。"""
try:
_post_json({"jsonrpc": "2.0", "method": method, "params": params}, timeout=15)
except (urllib.error.HTTPError, urllib.error.URLError, TimeoutError) as e:
print(f"[http] 通知失敗(継続): {e}")
def initialize():
"""セッションを新規に張る。古いIDは外して送らない。"""
global session_id
session_id = None
rpc("initialize", {
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "batch", "version": "1.0"},
})
notify("notifications/initialized", {})
def rpc(method, params, timeout=30, retries=4):
"""1要求を送る。セッション喪失(400/404)は初期化から回復、レート系は指数バックオフ。"""
delay = 1.0
for attempt in range(1, retries + 1):
try:
text = _post_json({"jsonrpc": "2.0", "id": attempt, "method": method, "params": params}, timeout)
return _extract_result(text, attempt)
except urllib.error.HTTPError as e:
if e.code in (400, 404) and method != "initialize":
print(f"[http] {e.code} → セッション喪失の疑い → 初期化からやり直し")
initialize()
continue
if e.code in (429, 503) and attempt < retries:
print(f"[http] {e.code} → {delay:.1f}秒待って再試行")
time.sleep(delay)
delay = min(delay * 2 + random.uniform(0, 0.3), 15.0)
continue
raise
except (urllib.error.URLError, TimeoutError) as e:
if attempt < retries:
print(f"[http] 通信断({type(e).__name__}) → {delay:.1f}秒待って再試行")
time.sleep(delay)
delay = min(delay * 2 + random.uniform(0, 0.3), 15.0)
continue
raise
raise RuntimeError(f"{retries}回の再試行でも通りませんでした: {method}")
if __name__ == "__main__":
initialize()
print(json.dumps(rpc("tools/call", {"name": "search", "arguments": {"query": "日誌 2026-10"}}),
ensure_ascii=False, indent=2))
404が返ってきたら「セッションが無効」と判断してsession_idを捨て、initializeからやり直します。サーバーの再起動・セッションの失効・デプロイ直後の接続切れがどれでも吸収できます。
2つのトランスポートの使い分け
| 観点 | stdio | Streamable HTTP |
|---|---|---|
| 寿命 | 子プロセスと共存。プロセスが死んだら一緒に止まる | サーバーは常駐。クライアントの寿命と独立 |
| 復旧 | プロセス再起動＋initializeのやり直しが必要 | セッションIDを払い直せば回復 |
| に向く | 手元ツール・数秒〜数分で終わるスクリプト | 夜間バッチ・長時間の複数ジョブ |
| 実装コスト | 標準ライブラリだけで完結できる | HTTPクライアント1個だが、セッション管理の理解が要る |
私の判断基準は「ジョブの寿命をプロセスと共有するか」です。共有するなら再接続層を必ず入れるか、HTTPに寄せるかの2択になります。
まとめ
-
子プロセスは夜間に静かに死ぬ前提で設計する。stderrは必ず保存し、死因を残す（直し方1の
err_buf） - 再接続は「プロセス再起動→initialize→同じ要求のやり直し」まで1本の関数にまとめる。MCPエラー応答（RuntimeError）は再試行しない理由とセットで設計する
- 長時間ジョブはStreamable HTTPに寄せる。寿命の共有が消え、残る失敗はセッション喪失だけになり、喪失時は初期化からやり直せば回復する
直し方1のクライアントはコピペで動き、デモサーバーの自爆も再現します。夜間にエージェントが沈黙して朝に慌てる日々を、今日で終わりにしてください。
参考書籍: Model Context Protocol実践入門 — エージェントとツールをつなぐ設計の教科書（Kindle・Kindle Unlimited読み放題対象）（葉山悠希 著）