応答が返らないMCP呼び出しに、無人の定期実行が丸ごと飲み込まれた日 — 自前のデッドラインで止め時を握る

import asyncio
import enum
from dataclasses import dataclass
from typing import Any, Awaitable, Callable
 
 
class Outcome(enum.Enum):
    OK = "ok"
    DEADLINE = "deadline"   # 自前デッドラインで打ち切った（ハング相当）
    ERROR = "error"         # サーバーが明示的に失敗を返した
 
 
@dataclass
class CallResult:
    outcome: Outcome
    value: Any = None
    detail: str = ""
    elapsed: float = 0.0
 
 
async def call_with_deadline(
    fn: Callable[[], Awaitable[Any]],
    deadline_s: float,
) -> CallResult:
    """MCPツール呼び出しを自前デッドラインで包み、結果を3分類で返す。
 
    deadline_s はプラットフォーム既定の打ち切りより内側に取ること。
    """
    loop = asyncio.get_event_loop()
    start = loop.time()
    try:
        value = await asyncio.wait_for(fn(), timeout=deadline_s)
        return CallResult(Outcome.OK, value=value, elapsed=loop.time() - start)
    except asyncio.TimeoutError:
        # 応答が返らないまま自前デッドラインに到達した
        return CallResult(
            Outcome.DEADLINE,
            detail=f"no response within {deadline_s}s",
            elapsed=loop.time() - start,
        )
    except Exception as e:  # noqa: BLE001 — 失敗の種類は呼び出し側で判断する
        return CallResult(
            Outcome.ERROR,
            detail=f"{type(e).__name__}: {e}",
            elapsed=loop.time() - start,
        )

import random
 
 
async def call_with_retry(
    fn: Callable[[], Awaitable[Any]],
    deadline_s: float,
    max_attempts: int = 3,
    base_backoff_s: float = 2.0,
) -> CallResult:
    """3分類の結果に応じて再試行する。冪等な呼び出しにのみ使うこと。"""
    last = CallResult(Outcome.ERROR, detail="not attempted")
    for attempt in range(1, max_attempts + 1):
        last = await call_with_deadline(fn, deadline_s)
        if last.outcome is Outcome.OK:
            return last
        if attempt == max_attempts:
            break
        # DEADLINE は相手の状態が不明なので、間隔を広めに取る
        factor = 2.0 if last.outcome is Outcome.DEADLINE else 1.0
        backoff = base_backoff_s * (2 ** (attempt - 1)) * factor
        backoff += random.uniform(0, backoff)  # フルジッター
        await asyncio.sleep(backoff)
    return last

import time
 
 
class ConnectorBreaker:
    """コネクタ単位の簡易サーキットブレーカ。
 
    連続失敗が閾値に達したら open にし、cooldown を過ぎるまで即スキップする。
    """
 
    def __init__(self, fail_threshold: int = 2, cooldown_s: float = 120.0):
        self.fail_threshold = fail_threshold
        self.cooldown_s = cooldown_s
        self._fails = 0
        self._opened_at = 0.0
 
    def is_open(self) -> bool:
        if self._fails < self.fail_threshold:
            return False
        if time.monotonic() - self._opened_at >= self.cooldown_s:
            self._fails = 0  # half-open: 一度だけ試させる
            return False
        return True
 
    def record(self, result: CallResult) -> None:
        if result.outcome is Outcome.OK:
            self._fails = 0
            return
        self._fails += 1
        if self._fails == self.fail_threshold:
            self._opened_at = time.monotonic()
 
 
async def guarded_call(
    breaker: ConnectorBreaker,
    fn: Callable[[], Awaitable[Any]],
    deadline_s: float,
) -> CallResult:
    if breaker.is_open():
        return CallResult(Outcome.ERROR, detail="breaker open: skipped")
    result = await call_with_retry(fn, deadline_s)
    breaker.record(result)
    return result