0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

MCPサーバーのエラーハンドリング — AIが自力で復帰する5つの設計パターン

0
Posted at

はじめに

自作MCPサーバーをClaude Codeに繋いだ直後、「ツールが1回エラーを返しただけで、その後の会話が丸ごと使えなくなる」症状にハマりました。原因は単純です。私はエラーを例外のまま投げていた。クライアント側にはスタックトレースが延々と返り、AIは「何が起きたか」も「次に何をすればいいか」も判断できないまま、同じ引数で同じツールを呼び続けました。

この記事では、エラーをAIが読んで次の手を選べる形で返す設計を、動くコード付きで書きます。ポイントは「例外を減らす」ことではなく、エラーを分類して、再試行の可否と具体的な次の一手を添えて返すことです。

  • 前提: Python 3.10+ / pip install "mcp[cli]"
  • ゴール: ツールが失敗しても、AIが自力で復帰できるサーバーにする

なぜ例外を投げるとAIが壊れるのか

MCPのツール呼び出しは JSON-RPC の tools/call で、成功も失敗も同じ結果オブジェクトとして返せます。仕様上は isError: true を立てて内容を返すのが正式なエラー表現です。

ところが FastMCP で raise すると、返る内容は Python の例外メッセージ(トレースバック込み)になります。人間が読む分には情報量がありますが、**AIにとっては「読んでも次の行動が決められない文字列」**です。結果、次のどちらかが起きます。

  1. 同じ引数で同じツールを再実行する(リトライの無限ループ)
  2. 会話全体を諦め、ツールを使わない回答に切り替える

私のケースは1でした。3回リトライした末に「在庫を確認できませんでした」と、利用者に丸投げして終わりました。

エラー設計の5パターン比較

# パターン 実装 AIが次にできること 向く場面 落とし穴
1 例外をそのまま投げる raise ほぼ何もできない 開発中のデバッグ 会話が壊れる・リトライ暴走
2 構造化して返す {"ok": false, "code": "NOT_FOUND"} 何が失敗したか理解する すべてのツール 情報が乏しいと結局リトライ
3 retryable を明示 上記 + 真偽値 再試行すべきか判断できる 一時障害(レート制限等) 恒久エラーに true を付けると暴走
4 suggested_action を添える 上記 + 文章 引数を直して再試行できる 入力ミス・形式エラー ヒントが曖昧だと無意味
5 サーキットブレーカを入れる 連続失敗で一時遮断 諦めて利用者に報告できる 外部API依存のツール 冷却が短いと素通りする

結論: 1は使いません。実務では 2+3+4 を基本にし、外部APIを叩くツールにだけ 5 を足すのが最小構成でした。5つ全部を最初から作る必要はありません。

実装(コピペで動く完全形)

1. AIに読ませるためのエラー型

まず「例外として投げない」ための入れ物を用意します。例外を投げる代わりに、この型で結果として返すのが要点です。

# errors.py
import json


class ToolError(Exception):
    """AIに『読ませる』ためのエラー。

    例外として送出せず、ツールの戻り値として返す。
    """

    def __init__(self, code, message, retryable=False, action=None):
        self.code = code
        self.message = message
        self.retryable = retryable
        self.action = action
        super().__init__(message)

    def to_payload(self):
        payload = {
            "ok": False,
            "code": self.code,
            "message": self.message,
            "retryable": self.retryable,
        }
        if self.action:
            payload["suggested_action"] = self.action
        return payload

    def to_json(self):
        return json.dumps(self.to_payload(), ensure_ascii=False)

2. 例外を「結果」に変換するデコレータ

ツール関数をこのデコレータで包むと、どんな例外も JSON 文字列として返るようになります。MCPのペイロードに載るのは常に正常な結果なので、会話は壊れません。

# safe_tool.py
import functools
import json
from errors import ToolError


def safe_tool(fn):
    """ツール関数を『絶対に例外を投げない』形に包む。"""

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        try:
            return fn(*args, **kwargs)
        except ToolError as e:
            return e.to_json()
        except Exception as e:  # 想定外の例外も握る
            return json.dumps(
                {
                    "ok": False,
                    "code": "INTERNAL",
                    "message": f"{type(e).__name__}: {e}",
                    "retryable": False,
                    "suggested_action": (
                        "引数を減らして1回だけ再試行する。"
                        "それでも失敗するなら利用者に状況を報告する"
                    ),
                },
                ensure_ascii=False,
            )

    return wrapper

functools.wraps は必須です。これが無いと ツールの説明文(docstring)が失われ、AIがツールを選べなくなります。地味ですが、私はここで半日溶かしました。

3. サーキットブレーカ(外部APIを叩くツール用)

外部APIが落ちているときにリトライを繰り返すと、失敗を量産するだけです。連続失敗で一時的に遮断します。

# breaker.py
import time


class CircuitBreaker:
    """連続失敗が閾値を超えたら一定時間ツールを閉じる。"""

    def __init__(self, threshold=3, cooldown=30.0):
        self.threshold = threshold
        self.cooldown = cooldown
        self.fails = 0
        self.opened_at = 0.0

    def allow(self):
        if self.fails < self.threshold:
            return True
        if time.time() - self.opened_at > self.cooldown:
            self.fails = 0  # 半開状態にして1回だけ通す
            return True
        return False

    def record(self, ok):
        if ok:
            self.fails = 0
        else:
            self.fails += 1
            if self.fails == self.threshold:
                self.opened_at = time.time()

4. サーバー本体(3つを組み合わせた完全形)

# server.py
# 実行: pip install "mcp[cli]" && python server.py
import json

from mcp.server.fastmcp import FastMCP

from breaker import CircuitBreaker
from errors import ToolError
from safe_tool import safe_tool

mcp = FastMCP("inventory")

# 外部API(ここでは在庫サービス)への依存に breaker を1つ持たせる
_breaker = CircuitBreaker(threshold=3, cooldown=30.0)
_STOCK = {"A-123": 12, "B-777": 0}


def _fetch_stock(sku):
    """外部の在庫サービスを呼ぶ想定(今は dict で代用)。"""
    if not _breaker.allow():
        raise ToolError(
            "UPSTREAM_UNAVAILABLE",
            "在庫サービスへの接続が連続失敗のため、一時的に遮断しています。",
            retryable=True,
            action="30秒待ってから1回だけ再実行する。それでも遮断中なら利用者に報告する",
        )
    ok = sku in _STOCK
    _breaker.record(ok)
    if not ok:
        raise ToolError(
            "NOT_FOUND",
            f"SKU '{sku}' は在庫サービスに存在しません。",
            retryable=False,
            action="list_skus を先に呼び、存在するSKUを確認してから再実行する",
        )
    return _STOCK[sku]


@mcp.tool()
@safe_tool
def get_stock(sku: str) -> str:
    """SKUを指定して在庫数を返す。SKUは 'A-123' の形式。"""
    if not sku or "-" not in sku:
        raise ToolError(
            "INVALID_SKU",
            f"SKU '{sku}' の形式が不正です。",
            retryable=True,
            action="SKUを『英字1文字-数字3桁』(例: A-123)の形式に直して再実行する",
        )
    return json.dumps(
        {"ok": True, "sku": sku, "stock": _fetch_stock(sku)}, ensure_ascii=False
    )


@mcp.tool()
def list_skus() -> str:
    """在庫サービスに登録されているSKUの一覧を返す。"""
    return json.dumps({"ok": True, "skus": sorted(_STOCK)}, ensure_ascii=False)


if __name__ == "__main__":
    mcp.run()

これで、AIが受け取るのは常に次のような判断可能なJSONになります。

{"ok": false, "code": "INVALID_SKU", "message": "SKU 'a123' の形式が不正です。", "retryable": true, "suggested_action": "SKUを『英字1文字-数字3桁』(例: A-123)の形式に直して再実行する"}

retryablesuggested_action があるだけで、AIは「引数を直して1回だけ再試行する」という正しい行動を選べます。実際、私の環境では同じ失敗のリトライ回数が3回→1回に減りました

私が踏んだ失敗パターン2つ

失敗1: retryable=True を恒久エラーに付けた

存在しないSKU(NOT_FOUND)に retryable=True を付けていた時期がありました。AIは「再試行可能」と解釈し、存在しないSKUを5回探し続けました。恒久エラーは retryable=False が必須です。判断基準は「待てば直るか」の一点です。

失敗2: デコレータの順序を逆にした

@safe_tool@mcp.tool()外側に置くと、ツール登録の時点で包まれず、例外がそのまま外へ出ます。正しい順序は上記のコードのとおり、@mcp.tool() を外側、@safe_tool を内側です。エラーが減らないときは、まずここを疑ってください。

まとめ

MCPサーバーのエラーハンドリングは、次の順で導入するのが最短でした。

  1. 例外を投げないsafe_tool で包み、常にJSONを返す
  2. retryable を正しく設定する — 「待てば直るか」で決める。判断を誤るとリトライが暴走する
  3. suggested_action に具体的な次の一手を書く — 「再試行してください」では不十分。引数の直し方まで書く
  4. 外部API依存のツールにだけ サーキットブレーカ を足す

次に試すなら、エラー種別ごとの統計を取るのがおすすめです。code をログに出すだけで「どのツールが・どの理由で・何回失敗したか」が見え、suggested_action の文面を実際の失敗に合わせて改善できます。ただし code を増やしすぎるとAIが分類できなくなるので、最初は10種類以内に抑えるのが無難です。

読者特典(無料)

本記事のような実装パターンを横断的にまとめたチートシートと、Obsidian・Notionのテンプレート集を無料配布しています。8冊の内容から「何をどのツールでやるか」の判断チャートも含みます。

🎁 読者特典を受け取る(無料・メール登録)

📗 MCP実践入門 — AIエージェントを拡張するModel Context Protocol(Kindle・読み放題対象)

著者: 葉山悠希 — 書籍シリーズは Zenn / Amazon で公開中

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?