KEY INSIGHT
MCP errors follow JSON-RPC conventions with structured codes and messagesΓÇöconsistent error handling enables reliable clients and simplifies debugging.
JSON-RPC defines standard error codes. Map your exceptions to appropriate codes:
```python
from mcp.types import ErrorData
ERROR_CODES = {
"parse_error": (-32700, "Invalid JSON"),
"invalid_request": (-32600, "Malformed request"),
"method_not_found": (-32601, "Unknown method"),
"invalid_params": (-32602, "Invalid parameters"),
"internal_error": (-32603, "Server error"),
# Application-specific codes (>= -32000)
"tool_not_found": (-32001, "Tool unavailable"),
"validation_failed": (-32002, "Input validation failed"),
"permission_denied": (-32003, "Access denied"),
"rate_limited": (-32004, "Too many requests"),
}
def error_response(code: str, message: str, data: any = None) -> dict:
code_num, _ = ERROR_CODES.get(code, ERROR_CODES["internal_error"])
return {
"jsonrpc": "2.0",
"error": {
"code": code_num,
"message": message,
"data": data,
},
"id": None, # Keep id from request if available
}
```
Structured error responses help clients handle failures:
```python
class ToolError(Exception):
def __init__(
self,
message: str,
code: str = "internal_error",
details: dict = None,
):
super().__init__(message)
self.code = code
self.details = details or {}
def to_response(self, request_id: int = None) -> dict:
return {
"jsonrpc": "2.0",
"id": request_id,
"error": {
"code": ERROR_CODES[self.code][0],
"message": self.message,
"data": self.details,
}
}
# Usage in tools
@mcp.tool()
async def read_file(path: str) -> str:
if not Path(path).exists():
raise ToolError(
f"File not found: {path}",
code="tool_not_found",
details={"path": path, "attempted": "read"}
)
return Path(path).read_text()
```
Global exception handling prevents protocol violations:
```python
async def handle_request(request: dict) -> dict:
try:
# Process request
result = await dispatch(request)
return {"jsonrpc": "2.0", "id": request.get("id"), "result": result}
except ValidationError as e:
return error_response("invalid_params", str(e), e.errors)
except PermissionError as e:
return error_response("permission_denied", str(e))
except Exception as e:
logger.exception(f"Unhandled error in request {request.get('id')}")
return error_response(
"internal_error",
"An unexpected error occurred",
{"type": type(e).__name__}
)
```
Retry logic handles transient failures:
```python
import asyncio
async def retry_with_backoff(
func,
max_attempts: int = 3,
base_delay: float = 1.0,
max_delay: float = 30.0,
):
last_exception = None
for attempt in range(max_attempts):
try:
return await func()
except TransientError as e:
last_exception = e
if attempt < max_attempts - 1:
delay = min(base_delay * (2 ** attempt), max_delay)
await asyncio.sleep(delay)
raise last_exception
```
Error recovery strategies handle partial failures:
```python
async def batch_operation(items: list[dict]) -> dict:
results = []
errors = []
for item in items:
try:
result = await process_item(item)
results.append({"item": item["id"], "status": "success", "result": result})
except Exception as e:
errors.append({"item": item["id"], "status": "failed", "error": str(e)})
return {
"completed": len(results),
"failed": len(errors),
"results": results,
"errors": errors,
}
```