Error handling - Runpod Documentation

Robust error handling is essential for production Serverless endpoints. It prevents your workers from crashing silently and ensures that useful error messages are returned to the client, making debugging significantly easier.

Basic error handling

The simplest way to handle errors is to wrap your handler logic in a try...except block. This ensures that even if your logic fails, the worker remains stable and returns a readable error message.

import runpod

def handler(job):
    try:
        input = job["input"]

        # Replace process_input() with your own handler logic
        result = process_input(input)

        return {"output": result}
    except KeyError as e:
        return {"error": f"Missing required input: {str(e)}"}
    except Exception as e:
        return {"error": f"An error occurred: {str(e)}"}

runpod.serverless.start({"handler": handler})

Structured error responses

For more complex applications, you should return consistent error objects. This allows the client consuming your API to programmatically handle different types of errors, such as validation failures versus unexpected server errors.

import runpod
import traceback

def handler(job):
    try:
        # Validate input
        if "prompt" not in job.get("input", {}):
            return {
                "error": {
                    "type": "ValidationError",
                    "message": "Missing required field: prompt",
                    "details": "The 'prompt' field is required in the input object"
                }
            }
        
        prompt = job["input"]["prompt"]
        result = process_prompt(prompt)
        return {"output": result}
        
    except ValueError as e:
        return {
            "error": {
                "type": "ValueError",
                "message": str(e),
                "details": "Invalid input value provided"
            }
        }
    except Exception as e:
        # Log the full traceback for debugging
        print(f"Unexpected error: {traceback.format_exc()}")
        return {
            "error": {
                "type": "UnexpectedError",
                "message": "An unexpected error occurred",
                "details": str(e)
            }
        }

runpod.serverless.start({"handler": handler})

Timeout handling

You can also set an execution timeout in your endpoint settings to automatically terminate a job after a certain amount of time.

For long-running operations, you may want to implement timeout logic within your handler. This prevents a job from hanging indefinitely and consuming credits without producing a result.

import runpod
import signal

class TimeoutError(Exception):
    pass

def timeout_handler(signum, frame):
    raise TimeoutError("Operation timed out")

def handler(job):
    try:
        # Set a timeout (e.g., 60 seconds)
        signal.signal(signal.SIGALRM, timeout_handler)
        signal.alarm(60)
        
        # Your processing code here
        result = long_running_operation(job["input"])
        
        # Cancel the timeout
        signal.alarm(0)
        
        return {"output": result}
        
    except TimeoutError:
        return {"error": "Request timed out after 60 seconds"}
    except Exception as e:
        return {"error": str(e)}

runpod.serverless.start({"handler": handler})