無応答の MCP サーバーでエージェントが止まる — 接続・一覧・呼び出しに別々のタイムアウトを敷く

import asyncio
from dataclasses import dataclass
 
 
@dataclass(frozen=True)
class TimeoutBudget:
    connect: float = 15.0      # 秒: 握手まで
    list_tools: float = 20.0   # 秒: ツール一覧
    call_tool: float = 120.0   # 秒: 1回の呼び出し
 
 
class BoundaryTimeout(Exception):
    """どの境界で時間切れになったかを保持する例外。"""
 
    def __init__(self, server: str, boundary: str, limit: float):
        self.server = server
        self.boundary = boundary
        self.limit = limit
        super().__init__(f"{server}: {boundary} timed out after {limit}s")
 
 
async def _guard(coro, *, server: str, boundary: str, limit: float):
    try:
        return await asyncio.wait_for(coro, timeout=limit)
    except asyncio.TimeoutError as exc:
        # 一律の TimeoutError では「どこで」止まったかが消える。
        # 境界名とサーバー名を載せ替えて投げ直すのが要点。
        raise BoundaryTimeout(server, boundary, limit) from exc
 
 
async def open_session(client, server: str, budget: TimeoutBudget):
    await _guard(client.connect(), server=server, boundary="connect", limit=budget.connect)
    tools = await _guard(
        client.list_tools(), server=server, boundary="list_tools", limit=budget.list_tools
    )
    return tools

import asyncio
import time
from dataclasses import dataclass, field
 
 
@dataclass
class Breaker:
    """連続失敗が閾値を超えたら、一定時間そのサーバーへの試行を止める。"""
 
    threshold: int = 3
    cooldown: float = 300.0  # 秒: 開いている間はスキップ
    fails: int = 0
    opened_at: float = 0.0
 
    def allow(self) -> bool:
        if self.fails < self.threshold:
            return True
        if time.monotonic() - self.opened_at >= self.cooldown:
            # クールダウンが明けたら半開き状態で 1 回だけ試す
            self.fails = self.threshold - 1
            return True
        return False
 
    def record(self, ok: bool) -> None:
        if ok:
            self.fails = 0
        else:
            self.fails += 1
            if self.fails == self.threshold:
                self.opened_at = time.monotonic()
 
 
class BreakerOpen(Exception):
    pass
 
 
async def with_retry(make_coro, *, server, boundary, limit, breaker,
                     attempts=3, base_delay=1.0, retryable=True):
    if not breaker.allow():
        raise BreakerOpen(f"{server}: breaker open, skipping {boundary}")
    last = None
    tries = attempts if retryable else 1
    for i in range(tries):
        try:
            result = await _guard(make_coro(), server=server, boundary=boundary, limit=limit)
            breaker.record(ok=True)
            return result
        except BoundaryTimeout as exc:
            last = exc
            breaker.record(ok=False)
            if i < tries - 1:
                # 指数バックオフ: 1s, 2s, 4s ... 無応答サーバーへの再突撃を間引く
                await asyncio.sleep(base_delay * (2 ** i))
    raise last

import json
import sys
 
 
def notify_failure(server: str, boundary: str, limit: float, run_id: str) -> None:
    """失敗時だけ 1 行の JSON を stderr に出す。上流の通知に拾わせる。"""
    payload = {
        "run_id": run_id,
        "server": server,
        "boundary": boundary,     # connect / list_tools / call_tool
        "limit_seconds": limit,
        "kind": "mcp_timeout",
    }
    print(json.dumps(payload, ensure_ascii=False), file=sys.stderr)
 
 
async def run_task(client, server, budget, breaker, run_id):
    try:
        await _guard(client.connect(), server=server, boundary="connect", limit=budget.connect)
        tools = await with_retry(
            client.list_tools, server=server, boundary="list_tools",
            limit=budget.list_tools, breaker=breaker,
        )
        return tools
    except BoundaryTimeout as exc:
        notify_failure(exc.server, exc.boundary, exc.limit, run_id)
        raise
    except BreakerOpen as exc:
        notify_failure(server, "breaker_open", 0.0, run_id)
        raise