Input Handling

Tasks can pause execution to request structured input from users. The input is validated against a JSON Schema before the task resumes.

Requesting Input

Define input schemas using Pydantic models:

from typing import Literal
from pydantic import BaseModel, Field
from redis_agent_kit import TaskManager

class DatabaseConfig(BaseModel):
    database: Literal["PostgreSQL", "MySQL", "Redis"]
    port: int = Field(default=5432, ge=1, le=65535)

task_manager = TaskManager(client)

await task_manager.request_input(
    task_id,
    prompt="Which database should I configure?",
    json_schema=DatabaseConfig.model_json_schema(),
)

After calling request_input: - Task status changes to AWAITING_INPUT - task.input_request.prompt contains the prompt - task.input_request.json_schema contains the JSON Schema

Submitting Input

Use AgentKit.submit_input() to validate and submit:

from redis_agent_kit import AgentKit

kit = AgentKit("redis://localhost:6379", redis_client=client)

# Valid input
await kit.submit_input(task_id, {"database": "Redis", "port": 6379})
# Task re-queues with status QUEUED

Validation

Input is validated against the JSON Schema. Invalid input raises ValueError:

# Missing required field
try:
    await kit.submit_input(task_id, {"port": 6379})
except ValueError as e:
    print(e)  # "'database' is a required property"

# Invalid type
try:
    await kit.submit_input(task_id, {"database": "Redis", "port": "not-a-number"})
except ValueError as e:
    print(e)  # "'not-a-number' is not of type 'integer'"

# Value out of range
try:
    await kit.submit_input(task_id, {"database": "Redis", "port": 99999})
except ValueError as e:
    print(e)  # "99999 is greater than the maximum of 65535"

# Invalid enum value
try:
    await kit.submit_input(task_id, {"database": "MongoDB"})
except ValueError as e:
    print(e)  # "'MongoDB' is not one of ['PostgreSQL', 'MySQL', 'Redis']"

Accessing Input in Handler

After input is submitted, the task resumes. Access the input:

async def my_handler(ctx):
    task = await ctx.kit.task_manager.get_task(ctx.task_id)

    if task and task.input_response:
        # User provided input
        database = task.input_response["database"]
        port = task.input_response.get("port", 5432)

        # Clear after use
        await ctx.kit.task_manager.clear_input(ctx.task_id)

        return {"configured": f"{database}:{port}"}
    else:
        # First run - request input
        class DbInput(BaseModel):
            database: str

        await ctx.kit.task_manager.request_input(
            ctx.task_id,
            prompt="Which database?",
            json_schema=DbInput.model_json_schema(),
        )
        return None  # Task pauses

Schema Examples

Simple Text

class TextInput(BaseModel):
    value: str = Field(min_length=1, max_length=100)

Boolean Confirmation

class Confirmation(BaseModel):
    confirmed: bool

Number with Range

class RangeInput(BaseModel):
    value: float = Field(ge=0, le=100)

Enum Selection

from typing import Literal

class Choice(BaseModel):
    option: Literal["option1", "option2", "option3"]

Complex Object

class ComplexInput(BaseModel):
    name: str
    count: int = Field(ge=1)
    tags: list[str] = []

Clearing Input

After processing, clear the input data:

await task_manager.clear_input(task_id)

This removes input_request and input_response from the task.

Task Status Flow

IN_PROGRESS → request_input() → AWAITING_INPUT
                                     ↓
                              submit_input()
                                     ↓
                                  QUEUED → Worker picks up → IN_PROGRESS

The same pause/resume pattern powers sub-tasking: a parent task that spawns children parks in AWAITING_CHILDREN and is re-enqueued when its children terminate. A child can also itself request input — the parent stays paused on AWAITING_CHILDREN until the child finishes (after its user input has been submitted). See the Sub-Tasks guide.

Reentrant Handlers with @side_effect

When a task resumes after input, the handler runs from the beginning. Any expensive operations (LLM calls, API requests) will repeat unless cached.

The @side_effect decorator makes functions idempotent by caching their completion status in Redis:

from redis_agent_kit import side_effect

@side_effect(store_result=True, redis_client=redis_client)
async def analyze_request(query: str) -> str:
    """This LLM call is cached across task re-runs."""
    return await llm.invoke(query)

async def my_handler(ctx):
    task = await ctx.kit.task_manager.get_task(ctx.task_id)

    # First run: calls LLM, caches result
    # Resume: returns cached result immediately
    analysis = await analyze_request(ctx.message)

    if task and task.input_response:
        # User provided confirmation
        return {"result": analysis}
    else:
        await ctx.kit.task_manager.request_input(
            ctx.task_id,
            prompt=f"Proceed with: {analysis}?",
            json_schema={"type": "object", "properties": {"confirm": {"type": "boolean"}}},
        )
        return None

Key Options

• store_result=True - Cache the return value (required if you need the result on resume)
• ttl=timedelta(hours=1) - How long to cache (default: 1 hour)
• key="custom-key" - Manual cache key instead of auto-generated

Key Generation

By default, keys are generated from: - Function source code hash (changes invalidate cache) - Function name - Input arguments

Control with key_policy:

from redis_agent_kit import FUNCTION_NAME, INPUTS

# Only use function name and inputs (ignore source changes)
@side_effect(key_policy=FUNCTION_NAME | INPUTS, redis_client=redis_client)
async def my_func(x: int):
    ...

Clearing Side Effects

from redis_agent_kit import clear_side_effects

# Clear all cached side effects
await clear_side_effects(redis_client, "all")

# Clear by function name
await clear_side_effects(redis_client, "analyze_request")

See examples/langgraph_interrupt/ for a complete example.