AI エージェントアプリケーションにおいて、MCP（Model Context Protocol）のエラー処理とリトライ機構は、安定したサービス提供の根幹を成します。本稿では、HolySheep AI を使用して、MCP 接続におけるエラー処理のベストプラクティスと実践的なリトライ機構の実装方法を詳しく解説します。
MCP エラー処理の重要性
MCP 接続では、ネットワーク不安定、API 制限、認証エラー等多种多様なエラーが発生します。適切なエラー処理を実装することで、アプリケーションの可用性を大幅に向上させることができます。HolySheep AI は50ミリ秒未満の低レイテンシを実現しており、効率的なリトライ処理が特に重要になります。
実践的なエラー処理パターン
1. 基本リトライ機構の実装
以下のコードは、指数バックオフ算法を用いた信頼性の高いリトライ機構の実装例です。HolySheep AI の API に対して安定して接続するための基盤となります。
import asyncio
import aiohttp
import time
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum
class RetryStrategy(Enum):
"""リトライ戦略の定義"""
IMMEDIATE = "immediate"
LINEAR = "linear"
EXPONENTIAL = "exponential"
@dataclass
class RetryConfig:
"""リトライ設定データクラス"""
max_retries: int = 3
base_delay: float = 1.0
max_delay: float = 60.0
strategy: RetryStrategy = RetryStrategy.EXPONENTIAL
retryable_status_codes: tuple = (429, 500, 502, 503, 504)
class MCPError(Exception):
"""MCP関連エラーの基底クラス"""
def __init__(self, message: str, code: Optional[str] = None, details: Optional[Dict] = None):
super().__init__(message)
self.code = code
self.details = details or {}
class HolySheepMCPClient:
"""
HolySheep AI MCP クライアント
ベースURL: https://api.holysheep.ai/v1
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str, retry_config: Optional[RetryConfig] = None):
self.api_key = api_key
self.retry_config = retry_config or RetryConfig()
self._session: Optional[aiohttp.ClientSession] = None
async def _get_session(self) -> aiohttp.ClientSession:
"""aiohttp セッションの遅延初期化"""
if self._session is None or self._session.closed:
self._session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self._session
def _calculate_delay(self, attempt: int) -> float:
"""リトライ間隔の計算"""
if self.retry_config.strategy == RetryStrategy.EXPONENTIAL:
delay = self.retry_config.base_delay * (2 ** (attempt - 1))
elif self.retry_config.strategy == RetryStrategy.LINEAR:
delay = self.retry_config.base_delay * attempt
else:
delay = self.retry_config.base_delay
return min(delay, self.retry_config.max_delay)
async def _execute_with_retry(
self,
method: str,
endpoint: str,
**kwargs
) -> Dict[str, Any]:
"""
リトライ機構付きリクエスト実行
HolySheep AI API へのリクエストを安定的に処理
"""
session = await self._get_session()
url = f"{self.BASE_URL}/{endpoint.lstrip('/')}"
last_exception = None
for attempt in range(1, self.retry_config.max_retries + 1):
try:
async with session.request(method, url, **kwargs) as response:
if response.status == 200:
return await response.json()
if response.status in self.retry_config.retryable_status_codes:
last_exception = MCPError(
message=f"リトライ可能なエラー: HTTP {response.status}",
code=f"HTTP_{response.status}",
details={"status": response.status, "attempt": attempt}
)
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
print(f"[Attempt {attempt}] {delay:.2f}秒後にリトライします...")
await asyncio.sleep(delay)
continue
error_body = await response.text()
raise MCPError(
message=f"リクエスト失敗: HTTP {response.status}",
code=f"HTTP_{response.status}",
details={"response": error_body, "status": response.status}
)
except aiohttp.ClientError as e:
last_exception = MCPError(
message=f"接続エラー: {str(e)}",
code="CONNECTION_ERROR",
details={"attempt": attempt, "error": str(e)}
)
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
print(f"[Attempt {attempt}] 接続エラー: {delay:.2f}秒後にリトライ...")
await asyncio.sleep(delay)
continue
except asyncio.TimeoutError:
last_exception = MCPError(
message="リクエストタイムアウト",
code="TIMEOUT",
details={"attempt": attempt}
)
if attempt < self.retry_config.max_retries:
delay = self._calculate_delay(attempt)
await asyncio.sleep(delay)
continue
raise last_exception
async def send_mcp_request(
self,
model: str,
messages: list,
max_tokens: int = 1000
) -> Dict[str, Any]:
"""
MCPリクエストの送信
HolySheep AI の高性能APIを使用
"""
payload = {
"model": model,
"messages": messages,
"max_tokens": max_tokens,
"stream": False
}
return await self._execute_with_retry(
"POST",
"chat/completions",
json=payload
)
async def close(self):
"""セッションのクリーンアップ"""
if self._session and not self._session.closed:
await self._session.close()2. MCP サーバー接続のエラー処理
MCP サーバーの接続においても同様のエラー処理が重要です。以下の例では、MCP プロトコル固有のエラータイプを適切に処理する方法を示します。
import asyncio
from typing import Callable, Optional, Any
from enum import Enum
import json
class MCPErrorType(Enum):
"""MCPエラータイプの定義"""
CONNECTION_FAILED = "connection_failed"
PROTOCOL_ERROR = "protocol_error"
TIMEOUT = "timeout"
AUTHENTICATION_FAILED = "authentication_failed"
RATE_LIMITED = "rate_limited"
SERVER_ERROR = "server_error"
INVALID_REQUEST = "invalid_request"
class MCPProtocolError(Exception):
"""MCPプロトコルエラーの基底クラス"""
def __init__(self, error_type: MCPErrorType, message: str, recoverable: bool = True):
super().__init__(message)
self.error_type = error_type
self.recoverable = recoverable
class ResilientMCPConnection:
"""
回復力のあるMCP接続クラス
HolySheep AI との接続を安定的に維持
"""
def __init__(
self,
api_key: str,
on_error: Optional