Error Handling
Robust error handling is essential for building reliable applications with ShotGrid MCP Server. This page covers best practices and patterns for handling errors at different levels of your application.
Types of Errors
When working with ShotGrid MCP Server, you may encounter several types of errors:
- ShotGrid API Errors: Errors returned by the ShotGrid API.
- Connection Errors: Network or authentication issues.
- Validation Errors: Invalid input data or parameters.
- Schema Errors: Issues with entity types or fields.
- MCP Protocol Errors: Errors in the MCP communication.
- Application Logic Errors: Errors in your application code.
Basic Error Handling
The simplest way to handle errors in tools is to use try-except blocks:
@server.tool()
def update_shot_status(shot_id: int, status: str) -> dict:
"""Update a shot's status with error handling."""
try:
# Validate the status
valid_statuses = ["wtg", "rdy", "ip", "cmpt", "fin"]
if status not in valid_statuses:
raise ValueError(
f"Invalid status: {status}. Must be one of: {', '.join(valid_statuses)}"
)
# Check if the shot exists
shot = server.connection.find_one("Shot", [["id", "is", shot_id]])
if not shot:
raise ValueError(f"Shot with ID {shot_id} not found")
# Update the shot
updated_shot = server.connection.update("Shot", shot_id, {"sg_status_list": status})
return {
"success": True,
"shot": updated_shot
}
except ValueError as e:
# Handle validation errors
return {
"success": False,
"error": "Validation Error",
"message": str(e)
}
except Exception as e:
# Handle unexpected errors
return {
"success": False,
"error": "Unexpected Error",
"message": str(e)
}
Raising Errors
For MCP tools, it’s often better to raise exceptions rather than returning error objects. The MCP protocol will automatically convert exceptions to appropriate error responses:
@server.tool()
def update_shot_status(shot_id: int, status: str) -> dict:
"""Update a shot's status with proper error raising."""
# Validate the status
valid_statuses = ["wtg", "rdy", "ip", "cmpt", "fin"]
if status not in valid_statuses:
raise ValueError(
f"Invalid status: {status}. Must be one of: {', '.join(valid_statuses)}"
)
# Check if the shot exists
shot = server.connection.find_one("Shot", [["id", "is", shot_id]])
if not shot:
raise ValueError(f"Shot with ID {shot_id} not found")
# Update the shot
updated_shot = server.connection.update("Shot", shot_id, {"sg_status_list": status})
return updated_shot
Custom Error Classes
For more structured error handling, define custom error classes:
# Define custom error classes
class ShotGridError(Exception):
"""Base class for ShotGrid-related errors."""
pass
class EntityNotFoundError(ShotGridError):
"""Raised when an entity is not found."""
def __init__(self, entity_type, entity_id):
self.entity_type = entity_type
self.entity_id = entity_id
message = f"{entity_type} with ID {entity_id} not found"
super().__init__(message)
class ValidationError(ShotGridError):
"""Raised when validation fails."""
pass
class PermissionError(ShotGridError):
"""Raised when permission is denied."""
pass
# Use custom error classes in tools
@server.tool()
def update_entity(entity_type: str, entity_id: int, data: dict) -> dict:
"""Update an entity with custom error handling."""
# Check if the entity exists
entity = server.connection.find_one(entity_type, [["id", "is", entity_id]])
if not entity:
raise EntityNotFoundError(entity_type, entity_id)
# Validate the data
if not data:
raise ValidationError("No data provided for update")
# Update the entity
try:
updated_entity = server.connection.update(entity_type, entity_id, data)
return updated_entity
except Exception as e:
if "Permission denied" in str(e):
raise PermissionError(f"Permission denied to update {entity_type} {entity_id}")
else:
# Re-raise other exceptions
raise
Handling ShotGrid API Errors
Catching Specific ShotGrid Errors
The ShotGrid API can raise various errors that you should handle specifically:
from shotgun_api3.shotgun import ShotgunError, Fault
@server.tool()
def create_entity_with_error_handling(entity_type: str, data: dict) -> dict:
"""Create an entity with specific ShotGrid error handling."""
try:
entity = server.connection.create(entity_type, data)
return {
"success": True,
"entity": entity
}
except Fault as e:
# Handle ShotGrid API faults
if "Entity of type" in str(e) and "cannot be created" in str(e):
return {
"success": False,
"error": "Creation Denied",
"message": f"Cannot create {entity_type}. You may not have permission."
}
elif "field" in str(e) and "does not exist" in str(e):
return {
"success": False,
"error": "Invalid Field",
"message": str(e)
}
else:
return {
"success": False,
"error": "ShotGrid Fault",
"message": str(e)
}
except ShotgunError as e:
# Handle other ShotGrid errors
return {
"success": False,
"error": "ShotGrid Error",
"message": str(e)
}
except Exception as e:
# Handle unexpected errors
return {
"success": False,
"error": "Unexpected Error",
"message": str(e)
}
Retry Logic for Transient Errors
Some ShotGrid API errors are transient and can be resolved by retrying:
import time
from shotgun_api3.shotgun import ShotgunError
@server.tool()
def find_with_retry(
entity_type: str,
filters: list,
fields: list,
max_retries: int = 3,
retry_delay: float = 1.0
) -> dict:
"""Find entities with retry logic for transient errors."""
retries = 0
last_error = None
while retries <= max_retries:
try:
# Attempt the operation
entities = server.connection.find(entity_type, filters, fields)
# If successful, return the results
return {
"success": True,
"entities": entities
}
except ShotgunError as e:
# Check if this is a transient error
error_str = str(e).lower()
is_transient = (
"timeout" in error_str or
"connection" in error_str or
"network" in error_str or
"temporarily" in error_str
)
if not is_transient:
# Non-transient error, don't retry
return {
"success": False,
"error": "ShotGrid Error",
"message": str(e)
}
# Store the error
last_error = str(e)
# Increment retry counter
retries += 1
if retries <= max_retries:
# Wait before retrying
time.sleep(retry_delay)
# Increase delay for next retry (exponential backoff)
retry_delay *= 2
else:
# Max retries reached, give up
break
# If we get here, all retries failed
return {
"success": False,
"error": "Max Retries Exceeded",
"message": last_error,
"retries": retries
}
Connection Pool Error Handling
The Connection Pool in ShotGrid MCP Server already handles many connection-related errors, but you can add additional error handling:
@server.tool()
async def safe_find_one(entity_type: str, entity_id: int) -> dict:
"""Find an entity with connection pool error handling."""
try:
# Get a connection from the pool
async with server.connection_pool.connection() as sg:
entity = sg.find_one(entity_type, [["id", "is", entity_id]])
if not entity:
return {
"success": False,
"error": "Not Found",
"message": f"{entity_type} with ID {entity_id} not found"
}
return {
"success": True,
"entity": entity
}
except Exception as e:
# The connection pool will handle connection errors,
# but we still need to handle other exceptions
return {
"success": False,
"error": "Error",
"message": str(e)
}
Validation Patterns
Validate input parameters before using them:
from typing import List, Optional, Literal
from pydantic import BaseModel, Field, validator
class CreateTaskRequest(BaseModel):
"""Request model for creating a task."""
content: str = Field(..., min_length=1, max_length=100)
entity_type: str
entity_id: int
status: Optional[Literal["wtg", "rdy", "ip", "cmpt", "fin"]] = "wtg"
@validator("entity_type")
def validate_entity_type(cls, v):
valid_types = ["Asset", "Shot", "Sequence"]
if v not in valid_types:
raise ValueError(f"entity_type must be one of: {', '.join(valid_types)}")
return v
@server.tool()
def create_task(request: CreateTaskRequest) -> dict:
"""Create a task with input validation using Pydantic."""
# Pydantic has already validated the input
# Check if the entity exists
entity = server.connection.find_one(
request.entity_type,
[["id", "is", request.entity_id]]
)
if not entity:
raise ValueError(f"{request.entity_type} with ID {request.entity_id} not found")
# Create the task
task_data = {
"content": request.content,
"entity": {"type": request.entity_type, "id": request.entity_id},
"sg_status_list": request.status
}
task = server.connection.create("Task", task_data)
return task
Schema Validation
Validate entity types and fields against the schema:
@server.tool()
def validate_entity_field(entity_type: str, field_name: str) -> dict:
"""Validate if a field exists for an entity type."""
schema = server.schema_loader.get_schema()
# Check if the entity type exists
if entity_type not in schema:
return {
"valid": False,
"error": "Invalid Entity Type",
"message": f"Entity type '{entity_type}' does not exist in the schema"
}
# Check if the field exists
if field_name not in schema[entity_type]:
return {
"valid": False,
"error": "Invalid Field",
"message": f"Field '{field_name}' does not exist for entity type '{entity_type}'"
}
# Get field information
field_info = schema[entity_type][field_name]
return {
"valid": True,
"field_type": field_info.get("data_type", {}).get("value"),
"editable": field_info.get("editable", {}).get("value", False)
}
Logging Errors
Implement logging to track errors:
import logging
# Configure logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.FileHandler("shotgrid_mcp.log"),
logging.StreamHandler()
]
)
logger = logging.getLogger("shotgrid_mcp")
@server.tool()
def create_entity_with_logging(entity_type: str, data: dict) -> dict:
"""Create an entity with error logging."""
try:
logger.info(f"Creating {entity_type} with data: {data}")
entity = server.connection.create(entity_type, data)
logger.info(f"Created {entity_type} with ID {entity['id']}")
return entity
except Exception as e:
logger.error(f"Error creating {entity_type}: {str(e)}", exc_info=True)
raise
Error Handling in Batch Operations
Batch operations require special error handling:
@server.tool()
def batch_operations_with_error_handling(operations: list) -> dict:
"""Perform batch operations with detailed error handling."""
try:
results = server.connection.batch(operations)
return {
"success": True,
"results": results
}
except Exception as e:
# For batch operations, we need to determine which operation failed
error_message = str(e)
# Try to identify the failed operation
failed_operation_index = None
# Look for patterns like "Error in operation 3:"
import re
match = re.search(r"Error in operation (\d+):", error_message)
if match:
failed_operation_index = int(match.group(1))
error_response = {
"success": False,
"error": "Batch Operation Failed",
"message": error_message
}
if failed_operation_index is not None:
error_response["failed_operation_index"] = failed_operation_index
if 0 <= failed_operation_index < len(operations):
error_response["failed_operation"] = operations[failed_operation_index]
return error_response
Graceful Degradation
Implement graceful degradation for non-critical features:
@server.tool()
def find_entities_with_thumbnails(entity_type: str, filters: list) -> dict:
"""Find entities with thumbnails, gracefully handling thumbnail errors."""
try:
# Find the entities
entities = server.connection.find(
entity_type,
filters,
["id", "code", "image"]
)
# Try to get thumbnails, but don't fail if they're not available
entities_with_thumbnails = []
for entity in entities:
try:
if entity.get("image"):
# Get thumbnail URL
thumbnail_url = server.connection.get_thumbnail_url(
entity_type,
entity["id"]
)
entity["thumbnail_url"] = thumbnail_url
else:
entity["thumbnail_url"] = None
except Exception as e:
# Log the error but continue
logger.warning(f"Error getting thumbnail for {entity_type} {entity['id']}: {e}")
entity["thumbnail_url"] = None
entities_with_thumbnails.append(entity)
return {
"success": True,
"entities": entities_with_thumbnails
}
except Exception as e:
# If the main query fails, that's a critical error
logger.error(f"Error finding {entity_type}: {e}")
raise
Client-Side Error Handling
When using the MCP client, handle errors appropriately:
from mcp.client import Client
from mcp.errors import MCPError, ToolExecutionError
async def handle_client_errors():
client = Client("http://localhost:8000")
try:
# Call a tool that might fail
result = await client.call_tool(
"update_shot",
{
"shot_id": 999999, # Non-existent shot
"status": "invalid_status"
}
)
print(f"Success: {result}")
except ToolExecutionError as e:
# Handle tool execution errors
print(f"Tool execution failed: {e.message}")
print(f"Error details: {e.details}")
except MCPError as e:
# Handle MCP protocol errors
print(f"MCP protocol error: {e}")
except Exception as e:
# Handle unexpected errors
print(f"Unexpected error: {e}")
Best Practices Summary
- Be Specific: Catch specific exceptions rather than using broad except blocks.
- Provide Context: Include relevant information in error messages.
- Use Custom Errors: Define custom error classes for different error types.
- Validate Early: Validate input data before performing operations.
- Log Errors: Implement logging to track errors.
- Retry Transient Errors: Implement retry logic for transient errors.
- Graceful Degradation: Allow non-critical features to fail gracefully.
- Consistent Error Format: Use a consistent format for error responses.
- Detailed Batch Errors: Provide detailed information for batch operation errors.
- Client-Side Handling: Implement proper error handling on the client side.
By following these patterns, you can build robust applications that handle errors gracefully and provide a better experience for your users.