When you start working with AI APIs, one of the most important skills to master is handling errors gracefully. Imagine you are building an application that relies on AI responses, and suddenly the API returns an error. Without proper error handling, your entire application crashes. This guide will walk you through everything you need to know about MCP (Model Context Protocol) error handling and retry mechanisms, using practical examples you can copy and run today.
If you are new to AI development, I recommend starting with a reliable provider. Sign up here for HolySheep AI, which offers competitive pricing at just $1 per dollar (saving you 85%+ compared to the industry average of ¥7.3), supports WeChat and Alipay payments, delivers responses in under 50ms latency, and provides free credits upon registration.
Understanding MCP Error Types
Before diving into code, let's understand what types of errors you might encounter when working with MCP. Think of errors as different types of traffic signals in your code—some are warnings (yellow), some are temporary blocks (red), and some are serious issues that require attention.
Common HTTP Status Codes in MCP
When you make a request to an MCP endpoint, the server responds with a status code. Here are the most common ones you will see:
- 200 OK: Your request was successful—the green light!
- 400 Bad Request: Something is wrong with your request format—check your parameters.
- 401 Unauthorized: Your API key is missing or invalid.
- 429 Too Many Requests: You have hit the rate limit—slow down and retry.
- 500 Internal Server Error: The server encountered an unexpected error—try again later.
- 503 Service Unavailable: The service is temporarily down—implement retry logic.
Screenshot hint: When you see an error in your terminal or console, look at the status code first. It usually appears at the top of the error message, often highlighted in red.
Setting Up Your Environment
Let us start from scratch. First, you need to install Python if you have not already. Download it from python.org and make sure to check the box that says "Add Python to PATH" during installation.
Next, install the requests library, which we will use to communicate with the MCP server:
pip install requestsNow, let us set up your environment variables. Create a new file called config.py and add your HolySheep API key:
# config.py
import os
HolySheep AI Configuration
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY"
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
Set environment variable for easy access
os.environ['HOLYSHEEP_API_KEY'] = HOLYSHEEP_API_KEY
os.environ['HOLYSHEEP_BASE_URL'] = HOLYSHEEP_BASE_URL
print("Configuration loaded successfully!")
print(f"Using base URL: {HOLYSHEEP_BASE_URL}")Replace YOUR_HOLYSHEEP_API_KEY with your actual key from the HolySheep dashboard. If you do not have an account yet, sign up here to get free credits to practice with.
Building Your First Error Handler
I remember the first time I encountered an API error in production—it was a 429 rate limit error that brought down my entire application at 3 AM. That experience taught me the importance of robust error handling. Let me share what I learned so you do not have to make the same mistakes.
Create a new file called error_handler.py and paste this complete error handling solution:
# error_handler.py
import requests
import time
import json
from typing import Optional, Dict, Any
class MCPToolError(Exception):
"""Custom exception for MCP tool errors"""
def __init__(self, message: str, status_code: int = None, retry_after: int = None):
self.message = message
self.status_code = status_code
self.retry_after = retry_after
super().__init__(self.message)
def handle_mcp_error(response: requests.Response) -> MCPToolError:
"""
Parse MCP error response and create appropriate exception.
Args:
response: The requests Response object
Returns:
MCPToolError with detailed information
"""
try:
error_data = response.json()
error_message = error_data.get('error', {}).get('message', 'Unknown error')
error_type = error_data.get('error', {}).get('type', 'unknown')
except json.JSONDecodeError:
error_message = response.text or 'Unknown error'
error_type = 'parse_error'
retry_after = None
if 'retry_after' in response.headers:
retry_after = int(response.headers['retry_after'])
elif error_type == 'rate_limit_error':
retry_after = 60 # Default 60 seconds for rate limits
return MCPToolError(
message=f"{error_type}: {error_message}",
status_code=response.status_code,
retry_after=retry_after
)
def call_mcp_tool(
base_url: str,
api_key: str,
tool_name: str,
parameters: Dict[str, Any],
max_retries: int = 3,
initial_delay: float = 1.0,
backoff_factor: float = 2.0
) -> Dict[str, Any]:
"""
Call an MCP tool with comprehensive error handling and retry logic.
Args:
base_url: The MCP server base URL
api_key: Your API key
tool_name: Name of the tool to call
parameters: Parameters to pass to the tool
max_retries: Maximum number of retry attempts
initial_delay: Initial delay between retries in seconds
backoff_factor: Multiplier for delay after each retry
Returns:
The tool response as a dictionary
Raises:
MCPToolError: If the request fails after all retries
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
endpoint = f"{base_url}/tools/{tool_name}"
delay = initial_delay
for attempt in range(max_retries):
try:
print(f"Attempt {attempt + 1}/{max_retries}: Calling {tool_name}...")
response = requests.post(
endpoint,
headers=headers,
json=parameters,
timeout=30
)
# Handle successful response
if response.status_code == 200:
print(f"Success! Received response in {response.elapsed.total_seconds():.3f}s")
return response.json()
# Handle rate limiting with retry
if response.status_code == 429:
error = handle_mcp_error(response)
if attempt < max_retries - 1:
wait_time = error.retry_after or delay
print(f"Rate limited. Waiting {wait_time} seconds before retry...")
time.sleep(wait_time)
delay *= backoff_factor
continue
else:
raise error
# Handle server errors with retry
if response.status_code >= 500:
if attempt < max_retries - 1:
print(f"Server error ({response.status_code}). Retrying in {delay} seconds...")
time.sleep(delay)
delay *= backoff_factor
continue
else:
error = handle_mcp_error(response)
raise error
# Handle client errors (4xx except 429) - no retry
error = handle_mcp_error(response)
raise error
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
print(f"Request timed out. Retrying in {delay} seconds...")
time.sleep(delay)
delay *= backoff_factor
continue
else:
raise MCPToolError("Request timed out after all retries", status_code=408)
except requests.exceptions.ConnectionError as e:
if attempt < max_retries - 1:
print(f"Connection error: {e}. Retrying in {delay} seconds...")
time.sleep(delay)
delay *= backoff_factor
continue
else:
raise MCPToolError(f"Failed to connect: {str(e)}", status_code=503)
except requests.exceptions.RequestException as e:
raise MCPToolError(f"Request failed: {str(e)}", status_code=0)
raise MCPToolError("Max retries exceeded", status_code=0)
Example usage
if __name__ == "__main__":
import os
api_key = os.environ.get('HOLYSHEEP_API_KEY')
base_url = os.environ.get('HOLYSHEEP_BASE_URL')
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
print("Please set your HOLYSHEEP_API_KEY in config.py")
else:
try:
result = call_mcp_tool(
base_url=base_url,
api_key=api_key,
tool_name="text-generation",
parameters={
"prompt": "Explain quantum computing in simple terms",
"max_tokens": 500,
"temperature": 0.7
}
)
print(f"Result: {json.dumps(result, indent=2)[:500]}")
except MCPToolError as e:
print(f"MCP Error: {e.message}")
if e.status_code:
print(f"Status Code: {e.status_code}")
if e.retry_after:
print(f"Retry After: {e.retry_after} seconds")Screenshot hint: When you run this script, you should see output like "Attempt 1/3: Calling text-generation..." followed by either "Success!" or error messages with specific status codes.
Implementing Exponential Backoff
Exponential backoff is a smart retry strategy where you wait progressively longer between each retry. Instead of waiting the same amount of time, you double (or multiply) the wait time after each failed attempt. This prevents overwhelming the server while giving it time to recover.
Here is an advanced implementation with jitter (random variation) to prevent thundering herd problems:
# retry_strategy.py
import random
import time
from functools import wraps
from typing import Callable, Any
class RetryStrategy:
"""
Configurable retry strategy with exponential backoff and jitter.
Attributes:
max_retries: Maximum number of retry attempts
base_delay: Initial delay in seconds
max_delay: Maximum delay cap in seconds
exponential_base: Multiplier for exponential growth
jitter: Whether to add random jitter (0.0 to 1.0)
"""
def __init__(
self,
max_retries: int = 5,
base_delay: float = 1.0,
max_delay: float = 60.0,
exponential_base: float = 2.0,
jitter: float = 0.1
):
self.max_retries = max_retries
self.base_delay = base_delay
self.max_delay = max_delay
self.exponential_base = exponential_base
self.jitter = jitter
def calculate_delay(self, attempt: int) -> float:
"""
Calculate delay for a given attempt number.
Args:
attempt: The current retry attempt (0-indexed)
Returns:
Delay in seconds
"""
# Exponential backoff: base_delay * (exponential_base ^ attempt)
delay = self.base_delay * (self.exponential_base ** attempt)
# Cap at maximum delay
delay = min(delay, self.max_delay)
# Add jitter if enabled
if self.jitter > 0:
jitter_range = delay * self.jitter
delay += random.uniform(-jitter_range, jitter_range)
return max(0.1, delay) # Minimum 100ms delay
def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any:
"""
Execute a function with retry logic.
Args:
func: The function to execute
*args: Positional arguments for the function
**kwargs: Keyword arguments for the function
Returns:
The function result
Raises:
The last exception if all retries fail
"""
last_exception = None
for attempt in range(self.max_retries + 1):
try:
result = func(*args, **kwargs)
if attempt > 0:
print(f"✓ Success after {attempt} retries")
return result
except Exception as e:
last_exception = e
if attempt < self.max_retries:
delay = self.calculate_delay(attempt)
print(f"✗ Attempt {attempt + 1} failed: {str(e)}")
print(f" Waiting {delay:.2f}s before retry...")
time.sleep(delay)
else:
print(f"✗ All {self.max_retries + 1} attempts failed")
raise last_exception
def with_retry(strategy: RetryStrategy = None):
"""
Decorator to add retry logic to any function.
Usage:
@with_retry(RetryStrategy(max_retries=3))
def my_function():
...
"""
if strategy is None:
strategy = RetryStrategy()
def decorator(func: Callable) -> Callable:
@wraps(func)
def wrapper(*args, **kwargs):
return strategy.execute_with_retry(func, *args, **kwargs)
return wrapper
return decorator
Advanced rate-limit-aware retry strategy
class RateLimitAwareRetry(RetryStrategy):
"""
Extended retry strategy that respects server-provided rate limits.
"""
def execute_with_retry(self, func: Callable, *args, **kwargs) -> Any:
last_exception = None
for attempt in range(self.max_retries + 1):
try:
result = func(*args, **kwargs)
if attempt > 0:
print(f"✓ Success after {attempt} retries")
return result
except Exception as e:
last_exception = e
error_str = str(e).lower()
# Check for rate limit indicators
retry_after = None
# Common rate limit indicators
if '429' in error_str or 'rate limit' in error_str:
# Try to extract retry-after value
import re
match = re.search(r'retry[-_\s]?after[:\s]+(\d+)', error_str)
if match:
retry_after = int(match.group(1))
else:
retry_after = 60 # Default 60 seconds
if attempt < self.max_retries:
if retry_after:
delay = retry_after
print(f"✗ Rate limited. Respecting server request to wait {delay}s")
else:
delay = self.calculate_delay(attempt)
print(f"✗ Attempt {attempt + 1} failed: {str(e)}")
print(f" Waiting {delay:.2f}s before retry...")
time.sleep(delay)
else:
print(f"✗ All {self.max_retries + 1} attempts failed")
raise last_exception
Example usage
if __name__ == "__main__":
import os
import requests
# Create a retry strategy
strategy = RateLimitAwareRetry(
max_retries=5,
base_delay=2.0,
max_delay=120.0,
jitter=0.2
)
# Test the retry logic with a mock API call
attempt_count = 0
def mock_api_call():
global attempt_count
attempt_count += 1
print(f" API call attempt {attempt_count}")
# Simulate random failures
if attempt_count < 3:
raise Exception("Simulated error: 429 Rate limit exceeded")
return {"status": "success", "attempts": attempt_count}
print("Testing retry strategy with simulated failures:")
result = strategy.execute_with_retry(mock_api_call)
print(f"Result: {result}")
# Reset counter
attempt_count = 0
print("\n" + "="*50)
print("Testing with @with_retry decorator:")
@with_retry(RetryStrategy(max_retries=3, base_delay=1.0))
def decorated_function():
global attempt_count
attempt_count += 1
print(f" Decorated function attempt {attempt_count}")
if attempt_count < 2:
raise Exception("Simulated transient error")
return {"message": "Success with decorator!", "attempts": attempt_count}
result = decorated_function()
print(f"Result: {result}")Handling Specific MCP Error Scenarios
Let us now look at real-world scenarios you will encounter and how to handle them properly.
Scenario 1: Authentication Failures
Authentication errors happen when your API key is missing, invalid, or expired. These should never be retried blindly—instead, log the issue and alert the developer immediately.
# auth_handler.py
import requests
from typing import Dict, Any, Optional
def handle_auth_error(
api_key: str,
base_url: str = "https://api.holysheep.ai/v1"
) -> Dict[str, Any]:
"""
Handle authentication-related errors.
Args:
api_key: Your API key
base_url: The base URL of the API
Returns:
Dictionary with authentication status and details
"""
result = {
"authenticated": False,
"error": None,
"error_type": None,
"suggestions": []
}
# Check if key is empty or placeholder
if not api_key or api_key == "YOUR_HOLYSHEEP_API_KEY":
result["error"] = "API key is missing or is a placeholder value"
result["error_type"] = "missing_key"
result["suggestions"] = [
"Generate an API key from your HolySheep dashboard",
"Copy the key exactly as shown (no extra spaces)",
"Store the key in environment variables, not in code"
]
return result
# Try to validate the key with a simple API call
try:
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Use the models endpoint to validate authentication
response = requests.get(
f"{base_url}/models",
headers=headers,
timeout=10
)
if response.status_code == 200:
result["authenticated"] = True
result["available_models"] = response.json().get("data", [])
return result
elif response.status_code == 401:
result["error"] = "Invalid API key"
result["error_type"] = "invalid_key"
result["suggestions"] = [
"Verify your API key is correct",
"Check if your account is active",
"Generate a new API key if needed"
]
return result
elif response.status_code == 403:
result["error"] = "API key does not have required permissions"
result["error_type"] = "insufficient_permissions"
result["suggestions"] = [
"Check if your plan includes API access",
"Upgrade your subscription if needed"
]
return result
else:
result["error"] = f"Unexpected response: {response.status_code}"
result["error_type"] = "unexpected_response"
return result
except requests.exceptions.Timeout:
result["error"] = "Authentication check timed out"
result["error_type"] = "timeout"
return result
except requests.exceptions.ConnectionError:
result["error"] = "Cannot connect to authentication service"
result["error_type"] = "connection_error"
result["suggestions"] = [
"Check your internet connection",
"Verify the API endpoint URL is correct"
]
return result
Practical usage example
def authenticated_mcp_request(
endpoint: str,
api_key: str,
method: str = "POST",
data: Optional[Dict] = None
) -> Dict[str, Any]:
"""
Make an authenticated MCP request with proper error handling.
"""
# First validate authentication
auth_result = handle_auth_error(api_key)
if not auth_result["authenticated"]:
return {
"success": False,
"error": auth_result["error"],
"error_type": auth_result["error_type"],
"suggestions": auth_result["suggestions"]
}
# Proceed with the actual request
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
try:
if method.upper() == "POST":
response = requests.post(endpoint, headers=headers, json=data, timeout=30)
else:
response = requests.get(endpoint, headers=headers, timeout=30)
return {
"success": True,
"status_code": response.status_code,
"data": response.json() if response.headers.get("content-type", "").startswith("application/json") else response.text
}
except Exception as e:
return {
"success": False,
"error": str(e),
"error_type": "request_failed"
}
Run examples
if __name__ == "__main__":
import os
# Test with invalid key
print("Testing with placeholder key:")
result = handle_auth_error("YOUR_HOLYSHEEP_API_KEY")
print(f"Authenticated: {result['authenticated']}")
print(f"Error: {result['error']}")
print(f"Suggestions: {result['suggestions']}")
print("\n" + "="*50)
print("Testing with real key:")
real_key = os.environ.get('HOLYSHEEP_API_KEY')
if real_key and real_key != "YOUR_HOLYSHEEP_API_KEY":
result = handle_auth_error(real_key)
print(f"Authenticated: {result['authenticated']}")
if result['authenticated']:
print(f"Available models: {len(result.get('available_models', []))}")
else:
print("No real API key found in environment. Skipping real test.")Scenario 2: Rate Limiting with Smart Wait
Rate limits are common when working with AI APIs. HolySheep AI offers competitive pricing that helps you stay well within limits while saving significantly compared to other providers. Here is how to handle rate limits gracefully:
# rate_limit_handler.py
import time
import requests
from datetime import datetime, timedelta
from collections import deque
from threading import Lock
class RateLimitHandler:
"""
Smart rate limit handler that tracks request patterns
and automatically throttles requests.
Features:
- Tracks requests per time window
- Calculates optimal delay between requests
- Respects server-provided rate limit headers
- Thread-safe for concurrent applications
"""
def __init__(
self,
requests_per_minute: int = 60,
burst_size: int = 10,
auto_adjust: bool = True
):
self.requests_per_minute = requests_per_minute
self.burst_size = burst_size
self.auto_adjust = auto_adjust
# Track request timestamps
self.request_history = deque(maxlen=1000)
self.lock = Lock()
# Server-reported limits
self.server_limit = None
self.server_remaining = None
self.server_reset_time = None
# Calculate optimal delay
self.min_delay = 60.0 / requests_per_minute
def record_request(self, response: requests.Response = None):
"""
Record a request and update rate limit tracking.
Args:
response: Optional response object to extract rate limit headers
"""
with self.lock:
now = datetime.now()
self.request_history.append(now)
# Extract server rate limit info from headers
if response is not None:
if 'x-ratelimit-limit' in response.headers:
self.server_limit = int(response.headers['x-ratelimit-limit'])
if 'x-ratelimit-remaining' in response.headers:
self.server_remaining = int(response.headers['x-ratelimit-remaining'])
if 'x-ratelimit-reset' in response.headers:
reset_timestamp = int(response.headers['x-ratelimit-reset'])
self.server_reset_time = datetime.fromtimestamp(reset_timestamp)
# Auto-adjust based on server feedback
if self.auto_adjust and self.server_remaining is not None:
if self.server_remaining < 10:
# Getting close to limit, slow down
self.requests_per_minute = min(
self.requests_per_minute,
self.server_remaining * 2
)
self.min_delay = 60.0 / max(self.requests_per_minute, 1)
def should_wait(self) -> tuple[bool, float]:
"""
Check if we should wait before making the next request.
Returns:
Tuple of (should_wait, wait_duration)
"""
with self.lock:
now = datetime.now()
cutoff = now - timedelta(minutes=1)
# Remove old requests from history
while self.request_history and self.request_history[0] < cutoff:
self.request_history.popleft()
# Count requests in the last minute
recent_requests = len(self.request_history)
# Check if we are at the rate limit
if recent_requests >= self.requests_per_minute:
# Calculate wait time until oldest request expires
oldest = self.request_history[0]
wait_time = (oldest + timedelta(minutes=1) - now).total_seconds()
return True, max(0.1, wait_time)
# Check server-reported remaining quota
if self.server_remaining is not None and self.server_remaining <= 0:
if self.server_reset_time:
wait_time = (self.server_reset_time - now).total_seconds()
return True, max(0.1, wait_time)
# Check burst limit
recent_burst = sum(
1 for ts in self.request_history
if ts > now - timedelta(seconds=10)
)
if recent_burst >= self.burst_size:
wait_time = 10.0 / self.burst_size
return True, wait_time
return False, 0.0
def wait_if_needed(self):
"""Wait if necessary to respect rate limits."""
should_wait, wait_time = self.should_wait()
if should_wait:
print(f"Rate limit handler: waiting {wait_time:.2f}s")
time.sleep(wait_time)
def make_request(
self,
url: str,
headers: dict,
method: str = "POST",
data: dict = None,
max_retries: int = 3
):
"""
Make a request with automatic rate limit handling.
"""
self.wait_if_needed()
for attempt in range(max_retries):
try:
if method.upper() == "POST":
response = requests.post(
url, headers=headers, json=data, timeout=30
)
else:
response = requests.get(url, headers=headers, timeout=30)
# Record this request
self.record_request(response)
# Handle rate limit response
if response.status_code == 429:
retry_after = int(response.headers.get('retry-after', 60))
if attempt < max_retries - 1:
print(f"Rate limited. Waiting {retry_after}s before retry...")
time.sleep(retry_after)
continue
else:
raise Exception(f"Rate limited after {max_retries} retries. Retry after: {retry_after}s")
return response
except Exception as e:
if attempt < max_retries - 1:
wait_time = 2 ** attempt
print(f"Request failed: {e}. Retrying in {wait_time}s...")
time.sleep(wait_time)
else:
raise
raise Exception("Max retries exceeded")
Usage example
if __name__ == "__main__":
handler = RateLimitHandler(requests_per_minute=30)
# Simulate making requests
for i in range(5):
print(f"Making request {i + 1}...")
# In real usage, you would call:
# response = handler.make_request(
# url="https://api.holysheep.ai/v1/chat/completions",
# headers={"Authorization": f"Bearer {api_key}"},
# data={"model": "gpt-4", "messages": [...]}
# )
# Simulate the response
print(f" Request {i + 1} completed at {datetime.now().strftime('%H:%M:%S')}")
# Simulate recording a response
class MockResponse:
headers = {
'x-ratelimit-limit': '60',
'x-ratelimit-remaining': '55',
'x-ratelimit-reset': str(int(time.time()) + 60)
}
handler.record_request(MockResponse())Creating a Production-Ready MCP Client
Now let us combine everything into a robust, production-ready MCP client that handles all edge cases:
# production_mcp_client.py
import time
import requests
import logging
from typing import Dict, Any, Optional, List, Callable
from dataclasses import dataclass
from enum import Enum
Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('MCPClient')
class ErrorSeverity(Enum):
"""Severity levels for error classification."""
LOW = "low" # Non-critical, can be retried immediately
MEDIUM = "medium" # Temporary issues, retry with backoff
HIGH = "high" # Configuration or auth issues, manual intervention needed
CRITICAL = "critical" # Data loss risk, stop immediately
@dataclass
class MCPError:
"""Structured error representation."""
message: str
status_code: Optional[int]
error_type: str
severity: ErrorSeverity
retryable: bool
retry_after: Optional[int] = None
suggestions: Optional[List[str]] = None
class ProductionMCPClient:
"""
Production-ready MCP client with comprehensive error handling,
retry logic, rate limiting, and monitoring.
Features:
- Automatic retry with exponential backoff
- Rate limiting with smart throttling
- Detailed error classification
- Request/response logging
- Circuit breaker pattern for cascading failures
"""
def __init__(
self,
api_key: str,
base_url: str = "https://api.holysheep.ai/v1",
max_retries: int = 5,
timeout: int = 30,
rate_limit_rpm: int = 60
):
self.api_key = api_key
self.base_url = base_url
self.max_retries = max_retries
self.timeout = timeout
self.rate_limiter = RateLimitHandler(requests_per_minute=rate_limit_rpm)
# Circuit breaker state
self.failure_count = 0
self.failure_threshold = 5
self.circuit_open = False
self.circuit_reset_time = None
# Monitoring
self.total_requests = 0
self.successful_requests = 0
self.failed_requests = 0
# Error callback for custom handling
self.error_callback: Optional[Callable[[MCPError], None]] = None
def _classify_error(self, status_code: int, response_text: str) -> MCPError:
"""Classify an error and determine severity and retryability."""
error_info = {
400: {
"type": "bad_request",
"severity": ErrorSeverity.HIGH,
"retryable": False,
"suggestions": [
"Check request parameters",
"Validate JSON format",
"Review API documentation"
]
},
401: {
"type": "authentication_error",
"severity": ErrorSeverity.CRITICAL,
"retryable": False,
"suggestions": [
"Verify API key is correct",
"Check if subscription is active",
"Generate new API key if compromised"
]
},
403: {
"type": "forbidden",
"severity": ErrorSeverity.CRITICAL,
"retryable": False,
"suggestions": [
"Check account permissions",
"Verify subscription plan includes this endpoint"
]
},
404: {
"type": "not_found",
"severity": ErrorSeverity.HIGH,
"retryable": False,
"suggestions": [
"Verify endpoint URL is correct",
"Check if resource still exists"
]
},
429: {
"type": "rate_limit",
"severity": ErrorSeverity.MEDIUM,
"retryable": True,
"suggestions": [
"Implement request queuing",
"Consider upgrading your plan"
]
},
500: {
"type": "server_error",
"severity": ErrorSeverity.MEDIUM,
"retryable": True,
"suggestions": [
"Server issue, retry later",
"Contact support if persistent"
]
},
502: {
"type": "bad_gateway",
"severity": ErrorSeverity.MEDIUM,
"retryable": True,
"suggestions": [
"Temporary server issue",
"Retry with exponential backoff"
]
},
503: {
"type": "service_unavailable",
"severity": ErrorSeverity.MEDIUM,
"retryable": True,
"suggestions": [
"Service temporarily down",
"Check status page for ongoing incidents"
]
}
}
info = error_info.get(status_code, {
"type": "unknown_error",
"severity": ErrorSeverity.MEDIUM,
"retryable": True,
"suggestions": ["Contact support with error details"]
})
return MCPError(
message=f"HTTP {status_code}: {response_text[:200]}",
status_code=status_code,
error_type=info["type"],
severity=info["severity"],
retryable=info["retryable"],
suggestions=info["suggestions"]
)
def _update_circuit_breaker(self, success: bool):
"""Update circuit breaker state."""
if success:
self.failure_count = 0
self.circuit_open = False
else:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
self.circuit_reset_time = time.time() + 60 # Reset after 60 seconds
logger.warning("Circuit breaker OPEN - too many failures")
def _check_circuit_breaker