File size: 2,067 Bytes
97cb846 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 | #!/usr/bin/env python3
"""
Simple Error Handling for MCPMark
==================================
Provides basic error standardization and retry logic.
"""
from typing import Optional
"""Retryable error detection via minimal substring matching (lower-case)."""
# Keep this list short and generic; aim to catch API/infrastructure issues only.
RETRYABLE_PATTERNS = {
"ratelimit", # e.g., RateLimitError, too many requests
# "connection", # connection refused/reset/error
"agent execution failed",
"unavailable", # service unavailable
# "execution timed out", # timeout
"internal server error", # 500s
"network error", # generic network issue
"quota", # budget/quota exceeded
# "llm provider not provided", # litellm error
# pipeline infra signals
"account balance",
"mcp network error",
"state duplication error",
"thought_signature",
"overloaded."
}
def is_retryable_error(error: str) -> bool:
"""Return True if the error string contains any retryable pattern."""
error_lower = str(error or "").lower()
return any(pattern in error_lower for pattern in RETRYABLE_PATTERNS)
def standardize_error_message(error: str, mcp_service: Optional[str] = None) -> str:
"""Standardize error messages for consistent reporting."""
error_str = str(error).strip()
# Common standardizations
if "timeout" in error_str.lower():
base_msg = "Operation timed out"
elif (
"connection refused" in error_str.lower() or "econnrefused" in error_str.lower()
):
base_msg = "Connection refused"
elif "not found" in error_str.lower():
base_msg = "Resource not found"
elif "already exists" in error_str.lower():
base_msg = "Resource already exists"
else:
# Return original message if no standardization applies
return error_str
# Add MCP service prefix if provided
if mcp_service:
return f"{mcp_service.title()} {base_msg}"
return base_msg
|