MCP サーバーに「チケットを作る」「通知を送る」といった書き込み系のツールを載せていると、ネットワークが一瞬切れたあとに「さっきの呼び出しは通ったのか」が分からなくなることがあります。作業は成功していて応答だけが失われたのなら、もう一度呼ぶと同じ作業が2回行われます。失敗していたのなら、呼び直さないと作業が抜け落ちます。クライアント側からは、このどちらなのかを見分けられません。
Claude Code の MCP ドキュメント（https://code.claude.com/docs/en/mcp 、ページ更新日 2026-10-28、2026-10-30 取得）は、この問題でサーバー作者が受け持つ役割をはっきり書いています。本稿では、その役割を果たすサーバーを空のディレクトリから組み立てます。
Claude Code は「2回呼んでも安全なツール」に印を付けるよう求めている
mark the tools that are safe to call twice
docs は MCP サーバー作者に、2回呼んでも安全なツールへ印を付けるよう求めています。接続が閉じた時点で保留中だった呼び出しを、Claude Code が再実行できるようにするためです。再実行してよいかどうかは Claude Code が決めるのではなく、ツールを公開するサーバー側の宣言で決まります。
印にはツールの annotations を使います。docs の説明では、idempotentHint は「状態を変え得るが、同じ引数でもう一度呼んでも安全」なツールを表し、readOnlyHint は「何も変えない」ツールを表します。readOnlyHint の付いたツールは、ほかの読み取り専用の呼び出しと並列に実行されるとも書かれています。ここから、書き込みのあるツールに誤って readOnlyHint を付けると並列実行の対象にもなる、と読めます（docs の記述をもとにした筆者の推論です）。
再実行は1回だけで、新しい接続の上で行われる
can repeat such a call to an HTTP server or a claude.ai connector once
再実行の対象は HTTP サーバーと claude.ai コネクタで、新しい接続の上で1回だけ行われます（2026-10-30 時点の docs）。接続まわりの再試行はほかにもあります。切断されたリモートサーバーに対しては、まず最大5回、1秒から始めて毎回倍にする指数バックオフで再接続を試みます。docs は、再試行が最大約30分続き、接続できた時点でそのサーバーのツールが再び使えるようになるとも説明しています。HTTP/SSE サーバーの初回接続が 5xx 応答・接続拒否・タイムアウトといった一時的なエラーで失敗した場合は、最大3回再試行します（いずれも 2026-10-30 時点）。
ここで押さえておきたいのは、再実行のリクエストが「別の接続」から届くことです。接続やセッションを手がかりに重複を判定するサーバーは、この再実行が同じ依頼だと気づけない可能性があります（docs の記述からの筆者の推論です）。
印の無いツールは「結果不明」のまま失敗する
connection closed before the tool call was answered
docs によると、Claude Code が再実行しない呼び出しは、このエラーで失敗することがあります。この時点では、サーバー側の処理が終わったのか、途中なのか、まだ始まっていないのかが分かりません。
反対に、本当は安全でないツールに idempotentHint を付けると、接続が切れるたびに同じ作業が2回走るおそれがあります。印は、2回呼ばれても安全であることを保証する実装と組み合わせて初めて役に立ちます。
この記事で手に入るもの
有料部分では、次のものをすべて載せています。
- 同じ引数での再実行を安全にする「冪等性の台帳」の全コード（Python 標準ライブラリと SQLite）
- idempotentHint / readOnlyHint を付けた MCP サーバー（Python SDK の FastMCP）
- 応答の消失、処理中の再送、キーの使い回し、プロセスの異常終了を再現するテスト
- Claude Code につないだ状態でサーバーを強制終了し、二重作成が起きないことを確かめる手順
- docs に書かれていない設計判断（キーの持たせ方、処理中に再送が届く場合、印を付けない判断）と既知の限界
最初の一歩: 公開しているツールを3つに仕分ける
実装に入る前に、手元のサーバーのツールを次の3つに分けてみてください。
- 何も変えないもの: readOnlyHint の候補
- 変更はするが、同じ引数で2回呼ばれても結果が1回分になるように作れるもの: idempotentHint の候補
- 2回目を止める手段が無いもの（下流の API に重複を防ぐ仕組みが無い送信など）: 印は付けず、状況を問い合わせるツールを別に用意する
2番目に入るツールを、実際に「2回呼ばれても1回分」にする作り方を、ここから先で扱います。
完成形と前提
作るのは、チケットを作成する MCP サーバー ticket-ledger です。ファイルは ledger.py（冪等性の台帳）、server.py（MCP サーバー）、test_retry.py（接続断の再現テスト）の3つです。
前提（2026-10-30 時点）は次のとおりです。
- Linux または macOS、Python 3.10 以上
- Claude Code CLI（本稿では MCP tasks を使わないため、tasks に必要な v2.1.304 以降という条件は必須ではありません）
- PyPI の mcp パッケージ。server.py は 1.x 系 FastMCP の API（FastMCP、ToolAnnotations）を前提に書いています。メジャーバージョンが変わって import 名が違う場合は、server.py の import を読み替えてください。ledger.py と test_retry.py は SDK に依存しません
- 本稿には実機で取得した端末出力を載せていません。各手順の成否は、終了コード、作られるファイル、DB の中身で判定します
設計の要点は3つです。
- 冪等性キー request_id をツールの引数に含め、Claude に生成させる
- キーの「確保」を SQLite の主キー制約によって1回だけ成功させ、後から来た呼び出しは完了を待って同じ結果を返す
- チケットの作成と「完了」の記録を、同じトランザクションで確定させる
手順1: 空のディレクトリから環境を作る
mkdir mcp-retry-ledger && cd mcp-retry-ledger
python3 -m venv .venv
. .venv/bin/activate
pip install "mcp[cli]"
pip show mcp
pip install の終了コードが 0 で、pip show が mcp パッケージの情報を返せば成功です。Version の行で、入ったバージョンを控えておいてください。
手順2: 「同じ引数なら安全」をサーバー側で成り立たせる ledger.py
the tool can change something, and calling it again with the same arguments is safe
idempotentHint の定義は「同じ引数でもう一度呼んでも安全」です。そこで、同じ引数の呼び出しが何回届いても作成は1回だけ行い、2回目以降には1回目と同じ結果を返す台帳を作ります。MCP から切り離したモジュールにしておくと、接続断の状況を SDK 抜きでテストできます。
"""request_id ごとに作成を1回だけ行う冪等性の台帳（標準ライブラリのみ）"""
import hashlib
import json
import os
import sqlite3
import time
import uuid
DB_PATH = os.environ.get("LEDGER_DB", "ledger.db")
WORK_SECONDS = float(os.environ.get("WORK_SECONDS", "0")) # 時間のかかる処理の模擬
WAIT_SECONDS = 20.0 # 実行中の呼び出しの完了を待つ上限
STALE_SECONDS = 120.0 # これより古い running は実行役が落ちたとみなす
BOOT_ID = uuid.uuid4().hex # このプロセスの識別子
class ConflictError(ValueError):
pass
def _connect():
conn = sqlite3.connect(DB_PATH, timeout=10, isolation_level=None)
conn.execute("PRAGMA journal_mode=WAL")
conn.executescript("""
CREATE TABLE IF NOT EXISTS requests(
request_id TEXT PRIMARY KEY,
args_hash TEXT NOT NULL,
status TEXT NOT NULL,
result TEXT,
owner TEXT NOT NULL,
claimed_at REAL NOT NULL);
CREATE TABLE IF NOT EXISTS tickets(
ticket_id TEXT PRIMARY KEY,
request_id TEXT NOT NULL UNIQUE,
title TEXT NOT NULL,
body TEXT NOT NULL);
""")
return conn
def _hash(title, body):
raw = json.dumps({"title": title, "body": body}, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode("utf-8")).hexdigest()
def _claim(conn, request_id, args_hash):
"""この呼び出しが実行役になれたら True を返す"""
now = time.time()
try:
conn.execute("INSERT INTO requests VALUES (?, ?, 'running', NULL, ?, ?)",
(request_id, args_hash, BOOT_ID, now))
return True
except sqlite3.IntegrityError:
pass
row = conn.execute("SELECT args_hash FROM requests WHERE request_id = ?",
(request_id,)).fetchone()
if row is None:
return False
if row[0] != args_hash:
raise ConflictError(f"request_id {request_id} は別の引数で使用済みです")
# 実行役が別プロセス（再起動前）か、古すぎる running だけを奪い直す
cur = conn.execute(
"UPDATE requests SET owner = ?, claimed_at = ? "
"WHERE request_id = ? AND status = 'running' AND (owner <> ? OR claimed_at < ?)",
(BOOT_ID, now, request_id, BOOT_ID, now - STALE_SECONDS))
return cur.rowcount == 1
def _do_work(conn, request_id, title, body):
time.sleep(WORK_SECONDS)
result = {"ticket_id": "T-" + uuid.uuid4().hex[:8], "title": title}
conn.execute("BEGIN IMMEDIATE")
try:
conn.execute("INSERT INTO tickets VALUES (?, ?, ?, ?)",
(result["ticket_id"], request_id, title, body))
conn.execute("UPDATE requests SET status = 'done', result = ? WHERE request_id = ?",
(json.dumps(result, ensure_ascii=False), request_id))
conn.execute("COMMIT")
except Exception:
conn.execute("ROLLBACK")
raise
return result
def _wait_done(conn, request_id):
deadline = time.time() + WAIT_SECONDS
while time.time() < deadline:
row = conn.execute("SELECT status, result FROM requests WHERE request_id = ?",
(request_id,)).fetchone()
if row is None:
return None
if row[0] == "done":
return json.loads(row[1])
time.sleep(0.2)
return None
def create_ticket(request_id, title, body):
if not isinstance(request_id, str) or not 1 <= len(request_id) <= 64:
raise ValueError("request_id は 1〜64 文字で指定してください")
conn = _connect()
try:
if _claim(conn, request_id, _hash(title, body)):
try:
return {**_do_work(conn, request_id, title, body), "replayed": False}
except sqlite3.IntegrityError:
pass # 別の実行役が先に確定させた。結果を読みに行く
except Exception:
conn.execute("DELETE FROM requests WHERE request_id = ? "
"AND status = 'running' AND owner = ?", (request_id, BOOT_ID))
raise
result = _wait_done(conn, request_id)
if result is None:
return {"request_id": request_id, "status": "in_progress",
"next": "find_request で状況を確認してから、同じ request_id で呼び直してください"}
return {**result, "replayed": True}
finally:
conn.close()
def find_request(request_id):
conn = _connect()
try:
row = conn.execute("SELECT status, result, owner FROM requests WHERE request_id = ?",
(request_id,)).fetchone()
finally:
conn.close()
if row is None:
return {"request_id": request_id, "status": "unknown"}
status, result, owner = row
if status == "running" and owner != BOOT_ID:
status = "abandoned" # 再起動前のプロセスが実行役のまま落ちた
return {"request_id": request_id, "status": status,
"result": json.loads(result) if result else None}
処理の流れは次のとおりです。
- requests に行を INSERT できた呼び出しだけが実行役になります。主キーが衝突したら、既存の行を確認します
- 既存の行と引数のハッシュが違えば ConflictError を投げます。同じ request_id を別の作成に使い回すのを止めるためです
- 実行役が別プロセス（再起動前のプロセス）のものか、STALE_SECONDS より古い running だけを奪い直します。UPDATE の条件で行を絞るので、奪い直せるのは1つの呼び出しだけです
- 実行役になれなかった呼び出しは WAIT_SECONDS まで完了を待ち、同じ結果に replayed: True を付けて返します
- 奪い直しが重なった場合でも、tickets.request_id の UNIQUE 制約によって2件目の INSERT は失敗します。失敗した側は、先に確定した結果を読みに行きます
手順3: server.py でツールに印を付ける
"""ticket-ledger: 接続断の後の再実行を安全にした MCP サーバー"""
import os
from anyio import to_thread
from mcp.server.fastmcp import FastMCP
from mcp.types import ToolAnnotations
import ledger
mcp = FastMCP("ticket-ledger", host="127.0.0.1", port=int(os.environ.get("PORT", "8000")))
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False,
idempotentHint=True))
async def create_ticket(request_id: str, title: str, body: str) -> dict:
"""チケットを1件作成する。
request_id: 新しいチケットごとに新しい一意な文字列（UUID など、1〜64文字）を生成する。
エラーや接続断の後にやり直すときは新しい ID を作らず、まず find_request で状況を確認し、
同じ request_id と同じ引数で呼び直す。status が in_progress の応答は未完了を意味する。"""
return await to_thread.run_sync(ledger.create_ticket, request_id, title, body)
@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True))
async def find_request(request_id: str) -> dict:
"""create_ticket に渡した request_id の状況を返す。
status は done / running / abandoned / unknown のいずれか。"""
return await to_thread.run_sync(ledger.find_request, request_id)
if __name__ == "__main__":
mcp.run(transport="streamable-http")
create_ticket には idempotentHint=True を、find_request には readOnlyHint=True を付けています。ツールを async にして台帳の処理をスレッドに逃がしているのは、1回目の呼び出しが待機している間も、新しい接続から届く再実行を受け付けられるようにするためです。同期関数のまま書いた場合にイベントループが塞がるかどうかは SDK の実装しだいなので、ここでは塞がらない書き方を選びました。
docstring も設計の一部です。FastMCP は関数の docstring をツールの説明として使うので、Claude が request_id をどう作り、どう使い回すかをここで指示しています。Claude Code の再実行は同じ呼び出しの繰り返しですが、エラーを受け取った Claude が自分の判断で呼び直すときに新しい ID を作ると、台帳はそれを別の作成として扱ってしまいます。
手順4: 接続断で起きる4つの状況をテストで再現する
"""接続断で起きる状況を、MCP を通さずに ledger で再現する"""
import os
import sqlite3
import tempfile
import threading
import time
os.environ["LEDGER_DB"] = os.path.join(tempfile.mkdtemp(), "ledger_test.db")
import ledger # noqa: E402 環境変数を設定してから読み込む
def count_tickets():
conn = sqlite3.connect(ledger.DB_PATH)
try:
return conn.execute("SELECT COUNT(*) FROM tickets").fetchone()[0]
finally:
conn.close()
# 1. 作成は終わったが応答が失われ、同じ引数で再実行された
r1 = ledger.create_ticket("req-1", "障害報告", "ログイン不可")
r2 = ledger.create_ticket("req-1", "障害報告", "ログイン不可")
assert r1["ticket_id"] == r2["ticket_id"]
assert (r1["replayed"], r2["replayed"]) == (False, True)
assert count_tickets() == 1
# 2. 1回目がまだ処理中のうちに、再実行が別の接続から届いた
ledger.WORK_SECONDS = 1.0
results = []
def call():
results.append(ledger.create_ticket("req-2", "請求修正", "金額誤り"))
first = threading.Thread(target=call)
second = threading.Thread(target=call)
first.start()
time.sleep(0.2)
second.start()
first.join()
second.join()
ledger.WORK_SECONDS = 0.0
assert len({r["ticket_id"] for r in results}) == 1
assert sorted(r["replayed"] for r in results) == [False, True]
assert count_tickets() == 2
# 3. 同じ request_id を別の作成に使い回した
try:
ledger.create_ticket("req-1", "別件", "別の本文")
except ledger.ConflictError:
pass
else:
raise AssertionError("ConflictError が出なかった")
assert count_tickets() == 2
# 4. 実行役のプロセスが作成前に落ち、running のまま残った
conn = sqlite3.connect(ledger.DB_PATH)
conn.execute("INSERT INTO requests VALUES (?, ?, 'running', NULL, 'old-process', ?)",
("req-3", ledger._hash("復旧確認", "再起動後"), time.time()))
conn.commit()
conn.close()
assert ledger.find_request("req-3")["status"] == "abandoned"
r3 = ledger.create_ticket("req-3", "復旧確認", "再起動後")
assert r3["replayed"] is False
assert ledger.find_request("req-3")["status"] == "done"
assert ledger.find_request("req-404")["status"] == "unknown"
assert count_tickets() == 3
print("ALL OK")
python test_retry.py
echo $?
成功の条件
- echo $? の結果が 0
- スクリプト末尾の print まで到達している（assert が1つでも失敗すると、そこで止まる）
- DB は tempfile の一時ディレクトリに作られ、作業ディレクトリには何も増えない
- 状況2の待機があるため、実行には1秒あまりかかる
失敗の見分け方
- AssertionError などのトレースバックが出て、終了コードが 0 以外になる
- トレースバックの行番号で、4つの状況のどれが崩れたかが分かる
手順5: Claude Code につなぎ、処理中にサーバーを落とす
ターミナル A でサーバーを起動します。
. .venv/bin/activate
WORK_SECONDS=8 python server.py
プロセスが終了せずに待ち受けを続けていれば成功です。ImportError ですぐに終了した場合は、手順1で控えた SDK のバージョンと server.py の import が合っているかを確認してください。
ターミナル B で登録します（URL のパスは FastMCP の streamable-http の既定値を前提にしています）。
claude mcp add --transport http ticket-ledger http://127.0.0.1:8000/mcp
claude mcp list
一覧に ticket-ledger が含まれ、接続失敗を示す表示になっていなければ次へ進めます。Claude Code を起動して、次のように依頼します。
ticket-ledger の create_ticket で、タイトル「障害報告」、本文「ログイン不可」のチケットを1件作ってください。
ツールが呼ばれてから8秒以内に、別のターミナルでサーバーを強制終了し、すぐにターミナル A で起動し直します。
pkill -9 -f "python server.py"
WORK_SECONDS=8 python server.py
その後は、次の2通りのどちらかになるはずです。
- Claude Code が新しい接続で呼び出しを1回再実行した場合: 新しいプロセスは BOOT_ID が違うので running を奪い直し、作成を完了させます
- 再実行されず、前述の接続断エラーになった場合: Claude に「find_request で状況を確認し、必要なら同じ request_id で create_ticket を呼び直して」と頼みます。find_request は abandoned を返し、呼び直しで作成が完了します
どちらになっても、最後に件数を確認します。
python -c "import sqlite3; c = sqlite3.connect('ledger.db'); print(c.execute('SELECT title, COUNT(*) FROM tickets GROUP BY title').fetchall())"
障害報告の件数が 1 なら成功です。2 になっていたら、Claude が新しい request_id で呼び直した可能性があります。requests テーブルに残っている request_id を見比べてください。なお、この手順で再現できるのは「プロセスの異常終了」です。「作成は済んだが応答だけが消えた」状況と「処理中に再送が届いた」状況は、手順4の状況1と状況2で確かめています。
docs に書かれていない設計判断
この節は docs の記述ではなく、筆者の設計上の意見です。根拠にした docs の記述は、その都度示します。
重複の判定を引数のハッシュだけに頼らず、キーを引数に入れる
引数のハッシュだけで重複を判定すると、同じタイトルのチケットを意図して2件作ろうとしたときに、2件目が黙って捨てられます。「やり直し」と「もう1件」を区別できるのは、呼び出す側だけです。そのため、区別の情報を request_id として引数に持たせ、ハッシュは「同じキーで別の内容が来た」ことの検出だけに使っています。
状態は接続にもプロセスにも持たせない
再実行は新しい接続の上で行われます。MCP tasks についても、docs は taskId を接続をまたいで問い合わせられるように保つよう求めています。セッション再開のあと、新しい接続でその ID が問い合わせられるからです。重複判定の状態も同じ理由で、接続ごとのメモリではなく永続ストアに置くべきだと考えます。複数インスタンス構成では、新しい接続が別のインスタンスに届く可能性もあります（推論です）。
再送は1回目が終わる前に届き得る
docs は、バックグラウンドタスクを停止してもサーバーが止まるとは限らず、作業の一部がすでに済んでいたり、最後まで完了したりすることがあると書いています。ここから、接続が切れたあとも1回目の処理がサーバー上で進み続け、その途中に再実行が届く状況を想定しました。台帳が「確保」と「完了待ち」に分かれているのはこのためです。待つ時間は無制限にできません。docs によると、サーバーごとの timeout はツール呼び出し1回あたりの壁時計時間の上限で、progress notification を送っても延びません。WAIT_SECONDS と処理時間の合計は、設定した timeout より短くしてください。上限を超えたら in_progress を返して、判断を Claude に戻す作りにしています。
副作用と完了の記録は1つのトランザクションに入れる。入れられないなら下流にキーを渡す
本稿のチケットは同じ SQLite に書くので、作成と完了の記録を1回のコミットで確定できます。外部 API を呼ぶ場合は、API 呼び出しと記録の間でプロセスが落ちると、再実行で二重送信になります。下流の API が重複防止のキーを受け取れるなら、request_id をそのまま渡します。受け取れないなら、そのツールには idempotentHint を付けず、最初の仕分けの3番目として扱うのが安全側の判断です。
結果が分からないときの出口を用意する
Before calling the tool again, find out from the server what was done
docs は、2回目の呼び出しで作業が繰り返されるおそれがあるとしたうえで、呼び直す前に、たとえばサーバーのほかのツールを使って何が済んだかを確かめるよう勧めています。find_request はそのためのツールです。ここで大事なのは、問い合わせに使うキーを「Claude がすでに知っている値」にすることです。サーバーが発行した ID は、失われた応答と一緒に Claude に届かなかった可能性があります。呼び出す側が渡した request_id で引ける作りにしておけば、応答が消えても状況をたどれます。idempotentHint を付けないツールにこそ、この問い合わせ用のツールを用意しておく価値があります。
応用: 2分を超える処理と MCP tasks
docs によると、メイン会話で2分を超えて実行中の MCP ツール呼び出しは、セッションを止めずにバックグラウンドタスクに移されます（2026-10-30 時点）。もっと長い処理には MCP tasks という選択肢がありますが、Claude Code CLI では public beta で、v2.1.304 以降が必要です（2026-10-30 時点）。本稿では実装していないので、要点だけ挙げます。
- taskId は接続をまたいで問い合わせられるようにしておく必要があります。本稿の requests テーブルと同じ永続ストアに置くのが自然です
- task の停止時、Claude Code は追跡をやめてサーバーにキャンセルを求めますが、サーバーが最後まで完了させる可能性は残ります
- tasks/cancel は、接続中に Claude や利用者がタスクを止めたときに送られ、Claude Code の終了時には送られません。リモートサーバー上の task は、終了後も動き続けます。キャンセル通知を後片付けのきっかけとして当てにしない設計が必要です
- リクエストが tasks の拡張を持つときだけ task で応答します。そうしないと、Claude にツールエラーが返ります
既知の限界
- BOOT_ID による奪い直しは、単一プロセスで動かすことを前提にしています。複数ワーカーや複数ホストで動かす場合は owner の条件を外し、STALE_SECONDS と共有 DB の UNIQUE 制約だけに頼ってください。その場合、STALE_SECONDS は最長の処理時間より長くする必要があります
- requests テーブルは増え続けます。古い行を消すなら、docs にある最大約30分の再試行期間（2026-10-30 時点）より十分長く残してください。筆者は1日以上を目安にしていますが、これは意見です
- docstring の指示（エラーのあとは同じ request_id で呼び直す）に、モデルが必ず従う保証はありません。手順5の件数確認のような、事後の検査と組み合わせてください
- _hash が対象にするのは title と body だけです。引数を増やしたら、_hash にも加えてください
- 例外がどのような形でツールエラーとして Claude に返るかや、annotations がどう送られるかは SDK の実装に依存します。使っているバージョンで確かめてください