Skip to main content
The Runpod SDK includes a built-in validation utility that ensures your handler receives data in the correct format before processing begins. Validating inputs early helps catch errors immediately and prevents your worker from crashing due to unexpected or malformed data types.

Import the validator

To use the validation features, import the validate function from the utils module:
from runpod.serverless.utils.rp_validator import validate

Define a schema

You define your validation rules using a dictionary where each key represents an expected input field. This schema dictates the data types, necessity, and constraints for the incoming data.
schema = {
    "text": {
        "type": str,
        "required": True,
    },
    "max_length": {
        "type": int,
        "required": False,
        "default": 100,
        "constraints": lambda x: x > 0,
    },
}
The schema supports several configuration keys:
  • type (required): Expected input type (e.g., str, int, float, bool).
  • required (default: False): Whether the field is required.
  • default (default: None): Default value if input is not provided.
  • constraints (optional): A lambda function that returns True or False to validate the value.

Validate input in your handler

When implementing validation in your handler, pass the input object and your schema to the validate function. The function returns a dictionary containing either an errors key or a validated_input key.
import runpod
from runpod.serverless.utils.rp_validator import validate

schema = {
    "text": {
        "type": str,
        "required": True,
    },
    "max_length": {
        "type": int,
        "required": False,
        "default": 100,
        "constraints": lambda x: x > 0,
    },
}


def handler(event):
    try:
        # Validate the input against the schema
        validated_input = validate(event["input"], schema)
        
        # Check for validation errors
        if "errors" in validated_input:
            return {"error": validated_input["errors"]}

        # Access the sanitized inputs
        text = validated_input["validated_input"]["text"]
        max_length = validated_input["validated_input"]["max_length"]

        result = text[:max_length]
        return {"output": result}
    except Exception as e:
        return {"error": str(e)}


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

Test the validator

You can test your validation logic locally without deploying. Save your handler code and run it via the command line with the --test_input flag.
python your_handler.py --test_input '{"input": {"text": "Hello, world!", "max_length": 5}}'
Alternatively, you can define your test case in a JSON file and pass it to the handler to simulate a real request.
test_input.json
{
  "input": {
    "text": "The quick brown fox jumps over the lazy dog",
    "max_length": 50
  }
}