Application Flow Control

This guide explains how to manage application flow and track progress using the Case class and related objects in your Supervaizer agent.

:::tip Mental model For the core concepts (Job, Case, Step) and how Supervaize relates to execution engines like n8n, see Core concepts. :::

Overview

info

This guide uses the CallAgent as the primary example.

See the Controller Setup Guide for initial setup instructions.

Application flow control allows you to:

Track individual tasks within a job using Cases
Update progress and status with CaseNodeUpdate
Request human input when decisions are needed
Return structured responses with JobResponse
Manage costs and errors throughout execution

Case Class and Methods

The Case class represents an individual task or unit of work within a job. Each case can track its own progress, costs, and results.

:::info Async event reporting Case and account methods that report to Supervaize are async. In async agents, use await Case.start(...), await case.update(...), await case.request_human_input(...), and await case.close(...).

For synchronous controller methods declared with is_async=False, use the explicit sync shims: Case.start_sync(...), case.update_sync(...), case.request_human_input_sync(...), and case.close_sync(...). :::

Case.start()

Creates a new case for tracking a specific task. This is a static method that returns a Case instance.

Parameters:

job_id (str, required): The ID of the parent job
account (Account, required): The account object from your controller setup
name (str, required): Descriptive name for the case
description (str, required): Detailed description of what this case represents
case_id (str, optional): Custom case identifier. If not provided, one will be generated

Returns: Case instance

Getting job_id from context:

The job_id is provided by the Supervaize platform when your agent method is called. Extract it from the context parameter in your method's kwargs:

# Get job_id from context
context_raw = kwargs.get("context")
if context_raw is None:
    raise ValueError("context is required in kwargs")

# Convert dict to JobContext if needed, otherwise use as-is
if isinstance(context_raw, dict):
    job_context = JobContext(**context_raw)
elif isinstance(context_raw, JobContext):
    job_context = context_raw
else:
    raise ValueError(f"context must be a dict or JobContext, got {type(context_raw)}")

job_instructions: JobInstructions = job_context.job_instructions
job_id = job_context.job_id

Explanation:

When Supervaize calls your agent method (e.g., job_start), it automatically passes a context parameter in kwargs
The context contains a JobContext object (or dict that can be converted to one) with job information
Extract job_id by accessing job_context.job_id after converting the context to a JobContext instance
This job_id uniquely identifies the job and is used to associate cases with their parent job

The context parameter is automatically provided by Supervaizer when your agent method is invoked. It contains a JobContext object with the job_id and other job-related information.

Example:

        case = await Case.start(
            job_id=job_id,
            account=account,
            name=f"Call to {phone_number}",
            description=f"Outbound call to {phone_number}",
            case_id=f"call_{phone_number}",
        )

When to create a case:

One case per phone number, email, or individual task
When you need to track multiple related but independent operations
When you want granular progress visibility in the Supervaize UI

case.update()

Updates a case with progress information, status changes, or intermediate results. Use this method throughout case execution to provide visibility into what's happening.

Parameters:

update (CaseNodeUpdate, required): The update object containing progress information

Example - Initial progress update:

                await case.update(
                    CaseNodeUpdate(
                        name=f"📞 Starting Outgoing Call (Try #{attempt})",
                        cost=0.0,
                        payload={
                            "attempt": attempt,
                            "call_id": call_id,
                        },
                        is_final=False,
                    )
                )

Example - Status updates during execution:

                        case.update(
                            CaseNodeUpdate(
                                name=f"#{iteration} {call_status}: {duration:.2f}s",
                                is_final=False,
                                payload=RetellCalls.status_fields(call_record),
                            )
                        )

Example - Final result update:

                case.update(
                    CaseNodeUpdate(
                        name=f"📄 Transcript for Outgoing Call (Try #{attempt})",
                        payload={
                            "attempt": attempt,
                            "call_id": call_id,
                            "call_status": call_status,
                            "transcript": call.transcript,
                            "cost so far": call.cost,
                            "expected_values": None,  # Will be populated later if available
                            "status_fields": status_fields,
                        },
                        is_final=False,
                    )
                )

Example - Success update:

                    case.update(
                        CaseNodeUpdate(
                            name="✅ Call Completed Successfully",
                            cost=call.cost,
                            payload={
                                "status": "success",
                                "call_id": call_id,
                                "call_status": call_status,
                                "duration_seconds": duration_seconds,
                                "transcript": str(call.transcript),
                                "expected_values": extracted_values or {},
                                "metadata": RetellCalls.metadata_fields(call_record),
                            },
                            is_final=False,
                        )
                    )

Example - Error update:

                case.update(
                    CaseNodeUpdate(
                        name=f"❌ Call Failed (Try #{attempt})",
                        cost=call_cost,
                        error=str(e),
                        payload={
                            "status": "failed",
                            "attempt": attempt,
                            "error": str(e),
                        },
                        is_final=False,
                    )
                )

The is_final parameter:

Set to False for intermediate updates (progress, status changes)
Set to True for final updates before closing the case
When True, this update represents the final state before case closure

case.patch_step()

Updates an existing step in-place instead of appending a new one. Use this when a later event should enrich or complete a step that was already sent — for example, adding an end time to a step that recorded a start time.

Parameters:

index (int, required): The index of the step to update. Must match the index of a previously sent step.
update (CaseNodeUpdate, required): The replacement update. Its index and upsert fields are set automatically.

How it works:

patch_step() sets upsert=True on the update before sending it to Studio. Studio then performs an update-or-create lookup by (case, index): if a step at that index already exists it is updated in-place; if not (e.g. the first call), it is created normally.

Example — recording both start and end time in a single step:

# On interview start — create the step
case.update(
    CaseNodeUpdate(
        name="Interview started",
        payload={"started_at": started_at},
        is_final=False,
    )
)
# index is auto-assigned, e.g. 5

# On interview end — patch the same step with the full session window
case.patch_step(
    index=5,
    update=CaseNodeUpdate(
        name="Interview session",
        payload={
            "started_at": started_at,
            "ended_at": ended_at,
            "duration_secs": duration_secs,
        },
        is_final=False,
    ),
)

When to use:

A step represents a time window (start + end), and you only want one step in the UI instead of two
A later event adds material information to an earlier step (e.g. a cost or result that wasn't known yet)
You want to avoid cluttering the step list with intermediate in-flight states

When NOT to use:

For independent progress updates — use case.update() to append a new step instead
When the new information is unrelated to the original step

case.request_human_input()

Requests human input during case execution. This pauses the case and waits for a human operator to provide input through the Supervaize UI.

Parameters:

update (CaseNodeUpdate, required): Update object containing the form definition in the payload with a supervaizer_form key
message (str, required): Human-readable message explaining what input is needed

Example:

            await human_pr_review_case.request_human_input(
                CaseNodeUpdate(
                    name="Human PR review",
                    cost=0.0,
                    payload={
                        "supervaizer_form": {
                            "question": f"Please review the PR : {pr_url} and approve or reject it. If approved,  it will be merged automatically, if rejected it will be closed.",
                            "answer": {
                                "fields": [
                                    {
                                        "name": "Approved",
                                        "description": "The PR has been approved",
                                        "type": bool,
                                        "field_type": "BooleanField",
                                        "required": False,
                                    },
                                    {
                                        "name": "Rejected",
                                        "description": "The PR has been rejected",
                                        "type": bool,
                                        "field_type": "BooleanField",
                                        "required": False,
                                    },
                                ],
                            },
                        }
                    },
                    is_final=False,
                ),
                "Please review the PR and approve it if it is ready to be merged.",
            )

Form structure: The supervaizer_form in the payload must contain:

question (str): The question or prompt to display to the human operator
answer (dict): Form definition with:
- fields (list): Array of field definitions following Django form field format
  - name (str): Field identifier
  - description (str): Field description
  - type: Python type (bool, str, int, etc.)
  - field_type (str): Django field type (BooleanField, CharField, IntegerField, etc.)
  - required (bool): Whether the field is required

When to use:

Approval workflows (PR reviews, content moderation)
Decision points requiring human judgment
Escalation scenarios
Quality checks before proceeding

Receiving Human Input ( human_answer Agent Method)

When request_human_input() is called, the case execution pauses and waits for human input. Once the human operator provides input through the Supervaize UI, the Supervaizer system calls a custom agent method to handle the response.

Important: receive_human_input() is not a method on the Case object. Instead, you must define a custom AgentMethod in your agent configuration that will be called when human input is received.

Setup:

Define a human_answer method in your supervaizer_control.py:

receive_human_input = AgentMethod(
    name="receive_human_input",
    method="your_module.handle_human_input",  # Your handler function
    is_async=False,
    fields=[]  # Human input is passed via context/payload, not fields _ TODO: verify
)

Add it to your agent's methods as a custom method:

Agent(
    # ... other config ...
    methods=AgentMethods(
        start=job_start,
        stop=job_stop,
        status=job_status,
        job_poll=poll_method,  # Optional: for manual polling
        human_answer= receive_human_input,
    )
)

Implement the handler function:

def handle_human_input(**kwargs) -> JobResponse:
    """Handle human input response."""
    context_raw = kwargs.get("context")
    job_context = JobContext(**context_raw)
    job_id = job_context.job_id

    # Human input is provided in the payload/fields
    fields = kwargs.get("fields", {})
    human_response = fields  # Contains the form field values

    # Get the case that requested input
    # (case_id or other identifier should be in context/payload)

    # Process the human input and continue case execution
    if human_response.get("Approved"):
        # Continue with approved operation
        pass
    elif human_response.get("Rejected"):
        # Handle rejection
        pass

    return JobResponse(
        job_id=job_id,
        status=EntityStatus.COMPLETED,
        message="Human input processed",
    )

How it works:

request_human_input() pauses the case and displays the form in the UI
When the human submits the form, Supervaizer calls your custom receive_human_input agent method
Your handler receives the form data in kwargs["fields"]
You can then continue the case execution based on the human's response

case.close()

Closes a case with final results and cost information. This should be called when the case is complete (successfully or with failure).

Parameters:

case_result (dict, required): Dictionary containing the final case results
final_cost (float, required): Total cost incurred for this case

Example - Success case:

                    case_result_dict = {
                        "status": "success",
                        "call_id": call_id,
                        "call_status": call_status,
                        "duration_seconds": duration_seconds,
                        "cost": call.cost,
                    }
                    await case.close(case_result=case_result_dict, final_cost=call.cost)

Example - Failure case:

                case_result_dict = {
                    "status": "failed",
                    "error": str(e),
                    "attempt": attempt,
                }
                await case.close(case_result=case_result_dict, final_cost=call_cost)

What to include in case_result:

status: Overall case status (e.g., "success", "failed")
Task-specific results (IDs, durations, extracted values, etc.)
Error information if the case failed
Any metadata relevant to the case outcome

When to close:

After successful completion of the task
After all retry attempts are exhausted
When an unrecoverable error occurs
Always close cases, even on failure, to provide final status

CaseNodeUpdate

CaseNodeUpdate represents a single update to a case, tracking progress, costs, and status information.

Fields

Field	Type	Default	Description
`index`	`int`	`None`	Step index within the case
`cost`	`float`	`None`	Cost incurred for this specific update
`name`	`str`	`None`	Descriptive name for this update (displayed in UI)
`payload`	`dict[str, Any]`	`None`	Arbitrary data associated with this update
`is_final`	`bool`	`False`	Whether this is a final update before case closure
`upsert`	`bool`	`False`	If `True`, Studio updates the existing step at the same index instead of creating a new one. Set automatically by `case.patch_step()`.
`error`	`str`	`None`	Error message if this update represents a failure
`scheduled_at`	`datetime`	`None`	When to execute this step (UTC). Defers execution.
`scheduled_method`	`str`	`None`	Dotted path to the agent method to call (e.g. `module.func`)
`scheduled_params`	`dict[str, Any]`	`None`	Parameters to pass to the scheduled method
`scheduled_status`	`str`	`None`	Execution state: `pending`, `executing`, `completed`, `failed`, `cancelled`

Usage Examples

Progress tracking:

                case.update(
                    CaseNodeUpdate(
                        name=f"📞 Starting Outgoing Call (Try #{attempt})",
                        cost=0.0,
                        payload={
                            "attempt": attempt,
                            "call_id": call_id,
                        },
                        is_final=False,
                    )
                )

Status updates:

                        case.update(
                            CaseNodeUpdate(
                                name=f"#{iteration} {call_status}: {duration:.2f}s",
                                is_final=False,
                                payload=RetellCalls.status_fields(call_record),
                            )
                        )

Human input requests:

                CaseNodeUpdate(
                    name="Human PR review",
                    cost=0.0,
                    payload={
                        "supervaizer_form": {
                            "question": f"Please review the PR : {pr_url} and approve or reject it. If approved,  it will be merged automatically, if rejected it will be closed.",
                            "answer": {
                                "fields": [
                                    {
                                        "name": "Approved",
                                        "description": "The PR has been approved",
                                        "type": bool,
                                        "field_type": "BooleanField",
                                        "required": False,
                                    },
                                    {
                                        "name": "Rejected",
                                        "description": "The PR has been rejected",
                                        "type": bool,
                                        "field_type": "BooleanField",
                                        "required": False,
                                    },
                                ],
                            },
                        }
                    },
                    is_final=False,
                ),

Final results:

                    case.update(
                        CaseNodeUpdate(
                            name="✅ Call Completed Successfully",
                            cost=call.cost,
                            payload={
                                "status": "success",
                                "call_id": call_id,
                                "call_status": call_status,
                                "duration_seconds": duration_seconds,
                                "transcript": str(call.transcript),
                                "expected_values": extracted_values or {},
                                "metadata": RetellCalls.metadata_fields(call_record),
                            },
                            is_final=False,
                        )
                    )

📖 See the model reference

CaseNodeUpdate Documentation

JobResponse

JobResponse is the return type for all agent methods (job_start, job_stop, job_status, job_poll, and custom methods). It provides structured information about the overall job execution status.

Fields

Field	Type	Default	Description
`job_id`	`str`	required	The ID of the job this response relates to
`status`	`EntityStatus`	required	Overall job status (COMPLETED, FAILED, RUNNING, CANCELLED, etc.)
`message`	`str`	required	Human-readable message describing the job status
`payload`	`dict[str, Any]`	`None`	Additional data about the job execution
`error_message`	`str`	`None`	to verify - Error message if job failed (may be `error` instead)
`error_traceback`	`str`	`None`	Full error traceback if job failed

Usage Examples

Success response:

    return JobResponse(
        job_id=job_id,
        status=overall_status,
        message=message,
        payload={
            "total_calls": len(phone_numbers),
            "success_count": success_count,
            "failure_count": failure_count,
            "total_cost": total_cost,
            "results": all_results,
        },
    )

Error response:

        return JobResponse(
            job_id=job_id if "job_id" in locals() else "unknown",
            status=EntityStatus.FAILED,
            message=f"Job failed: {str(e)}",
            payload={"error": str(e)},
            error=str(e),
        )

Stop response:

    return JobResponse(
        job_id=job_id,
        status=EntityStatus.CANCELLED,
        message="Call job stop requested",
        payload={"status": "stopped"},
    )

Status response:

    return JobResponse(
        job_id=job_id,
        status=EntityStatus.RUNNING,
        message="Call job status retrieved",
        payload={"status": "unknown"},
    )

When to Use JobResponse vs Case Updates

Use JobResponse for:

Overall job status (the job as a whole)
Aggregated results across multiple cases
Job-level errors or failures
Return value from agent methods

Use Case updates for:

Individual task progress within a job
Task-specific status and results
Granular cost tracking per task
Task-level errors and retries

Relationship:

A JobResponse represents the entire job execution
A Case represents an individual task within that job
One job can have multiple cases
Each case tracks its own progress and results
The final JobResponse typically aggregates results from all cases

📖 See the model reference

JobResponse Documentation

Manual Polling (job_poll)

Agents that depend on external events (webhooks from email services, telephony providers, etc.) can implement a job_poll method for manual status checking. This is especially useful for local development where webhooks aren't available.

When defined, the workbench shows a "Check for updates" button on active jobs. Clicking it calls the agent's poll handler, which checks external services and updates cases accordingly.

def my_agent_poll(**kwargs) -> JobResponse:
    """Check external services for updates."""
    job_id = kwargs.get("job_id", "")
    # Check your external service (email inbox, call status, etc.)
    # Update relevant cases with new information
    return JobResponse(
        job_id=job_id,
        status=EntityStatus.AWAITING,
        message="Found 2 new replies",
    )

methods=AgentMethods(
    job_start=start_method,
    job_stop=stop_method,
    job_poll=AgentMethod(
        name="poll",
        method="my_module.my_agent_poll",
        description="Check for external updates",
    ),
    human_answer=answer_method,
)

The poll method receives {"job_id": job_id} as kwargs and returns a JobResponse. The workbench refreshes the monitor after each poll.

Webhook + Poll Pattern: In production, use webhooks for real-time updates. In local/dev mode, the poll button provides the same functionality without needing a public URL. Your agent code should handle both paths.

Scheduled Steps

Scheduled steps let you defer execution of a case update to a future time. When a step has scheduled_at set, the background executor holds it until the scheduled time, then automatically calls the specified agent method.

This is useful for:

Time-based orchestration — schedule a follow-up call in 30 minutes
Retry with backoff — retry a failed operation after a delay
Workflow delays — insert a waiting period between steps

Scheduling a step

Add the scheduled_* fields to a CaseNodeUpdate:

from datetime import datetime, timedelta, timezone

case.update(
    CaseNodeUpdate(
        name="Follow-up call scheduled",
        scheduled_at=datetime.now(timezone.utc) + timedelta(minutes=30),
        scheduled_method="my_agent.follow_up_call",
        scheduled_params={"phone_number": "+1234567890", "case_id": case.case_id},
        is_final=False,
    )
)

scheduled_at: UTC datetime when the step should execute
scheduled_method: Dotted path to a callable (must be importable by the server)
scheduled_params: Dict of keyword arguments passed to the method
scheduled_status: Automatically set to pending — do not set this manually

How the executor works

The server starts a background asyncio loop on startup that:

Polls every 60 seconds for due steps (scheduled_at <= now() and scheduled_status == "pending")
Sets scheduled_status to executing
Dynamically imports and calls the scheduled_method with scheduled_params
Updates scheduled_status to completed on success or failed on error

Workbench controls

The workbench shows scheduled steps with visual status badges:

Status	Badge	Description
`pending`	Blue	Waiting for scheduled time, shows countdown
`executing`	Amber ⚡	Currently running
`completed`	Green ✅	Finished successfully
`failed`	Red ❌	Execution failed
`cancelled`	Gray	Cancelled by user or job stop

For pending steps, the workbench displays:

A countdown timer showing time until execution
An "Execute now" button to trigger immediately
A "Cancel" button to cancel the scheduled step

Cancellation on job stop

When a job is stopped, Case.cancel_scheduled_steps() automatically sets all pending scheduled steps to cancelled, preventing orphaned executions.

REST API for scheduled steps

Three endpoints are available for programmatic control:

Method	Endpoint	Description
`POST`	`/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/execute`	Execute a pending step immediately
`POST`	`/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/cancel`	Cancel a pending step
`PATCH`	`/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/schedule`	Reschedule to a new time (body: `{"scheduled_at": "ISO8601"}`)

Complete Example Flow

Here's a complete workflow showing how all the pieces work together, based on the CallAgent example:

def call_agent_start(**kwargs: Any) -> JobResponse:
    """Start calls to phone numbers - complete workflow example."""

    # 1. Extract job context and parameters
    fields = kwargs.get("fields", {})
    context_raw = kwargs.get("context")
    job_context = JobContext(**context_raw)
    job_id = job_context.job_id

    phone_numbers = [pn.strip() for pn in fields.get("phone_numbers", "").split(",") if pn.strip()]

    # 2. Initialize resources
    retell_client = RetellClient()
    handler = CallHandler(retell_client=retell_client)

    # 3. Process each phone number as a separate case
    all_results = []
    success_count = 0
    failure_count = 0

    for phone_number in phone_numbers:
        # 3a. Create a case for this phone number
        case = Case.start(
            job_id=job_id,
            account=account,
            name=f"Call to {phone_number}",
            description=f"Outbound call to {phone_number}",
            case_id=f"call_{phone_number}",
        )

        # 3b. Try calling with retries
        for attempt in range(1, max_retries + 1):
            try:
                # 3c. Update case with initial progress
                case.update(
                    CaseNodeUpdate(
                        name=f"📞 Starting Outgoing Call (Try #{attempt})",
                        cost=0.0,
                        payload={"attempt": attempt, "call_id": call_id},
                        is_final=False,
                    )
                )

                # 3d. Execute the call and update status
                # ... call execution logic ...

                # 3e. Update with intermediate status
                case.update(
                    CaseNodeUpdate(
                        name=f"#{iteration} {call_status}: {duration:.2f}s",
                        is_final=False,
                        payload=RetellCalls.status_fields(call_record),
                    )
                )

                # 3f. Update with final transcript
                case.update(
                    CaseNodeUpdate(
                        name=f"📄 Transcript for Outgoing Call (Try #{attempt})",
                        payload={
                            "transcript": call.transcript,
                            "cost so far": call.cost,
                        },
                        is_final=False,
                    )
                )

                # 3g. If successful, update and close case
                if call_status in ["ended", "completed"]:
                    case.update(
                        CaseNodeUpdate(
                            name="✅ Call Completed Successfully",
                            cost=call.cost,
                            payload={"status": "success", ...},
                            is_final=False,
                        )
                    )
                    case.close(
                        case_result={"status": "success", "call_id": call_id, ...},
                        final_cost=call.cost
                    )
                    success_count += 1
                    break

            except Exception as e:
                # 3h. On error, update and close case with failure
                case.update(
                    CaseNodeUpdate(
                        name=f"❌ Call Failed (Try #{attempt})",
                        cost=call_cost,
                        error=str(e),
                        payload={"status": "failed", "error": str(e)},
                        is_final=False,
                    )
                )
                case.close(
                    case_result={"status": "failed", "error": str(e)},
                    final_cost=call_cost
                )
                failure_count += 1

    # 4. Return final JobResponse with aggregated results
    return JobResponse(
        job_id=job_id,
        status=EntityStatus.COMPLETED if failure_count == 0 else EntityStatus.FAILED,
        message=f"Processed {len(phone_numbers)} phone number(s): {success_count} success, {failure_count} failed",
        payload={
            "total_calls": len(phone_numbers),
            "success_count": success_count,
            "failure_count": failure_count,
            "results": all_results,
        },
    )

Key patterns:

Job-level: Start with job context, return JobResponse
Case-level: Create a Case for each independent task
Progress: Use await case.update() with CaseNodeUpdate throughout execution
Completion: Always await case.close() with results, even on failure
Aggregation: Final JobResponse summarizes all cases

Best Practices

When to Create Multiple Cases vs Single Case

Create multiple cases when:

Processing multiple independent items (phone numbers, emails, files)
Each item can succeed or fail independently
You want granular visibility into each item's progress
Items can be processed in parallel or have different outcomes

Use a single case when:

All work is part of a single cohesive task
Steps are sequential and dependent
You don't need granular tracking of sub-tasks
The entire task succeeds or fails together

Cost Tracking Patterns

Track costs at multiple levels:

Per CaseNodeUpdate: Use the cost field for individual operation costs
Per case: Accumulate costs and pass final_cost to case.close()
Per job: Aggregate case costs in the final JobResponse payload

Example:

# Track cost per update
await case.update(CaseNodeUpdate(name="API call", cost=0.05, ...))

# Track total cost when closing
total_case_cost = sum(update.cost for update in case.updates)
await case.close(case_result={...}, final_cost=total_case_cost)

Error Handling Patterns

Always close cases, even on failure:

try:
    # ... execute task ...
    await case.close(case_result={"status": "success", ...}, final_cost=cost)
except Exception as e:
    await case.update(CaseNodeUpdate(name="Error", error=str(e), ...))
    await case.close(case_result={"status": "failed", "error": str(e)}, final_cost=cost)

Provide context in error updates:

case.update(
    CaseNodeUpdate(
        name="❌ Operation Failed",
        error=str(e),
        payload={
            "error": str(e),
            "attempt": attempt,
            "context": relevant_context,
        },
        is_final=False,
    )
)

Use retry patterns with case updates:

for attempt in range(1, max_retries + 1):
    try:
        case.update(CaseNodeUpdate(name=f"Attempt {attempt}", ...))
        # ... execute ...
        break
    except Exception as e:
        if attempt == max_retries:
            case.close(case_result={"status": "failed", ...}, final_cost=cost)

Human Input Workflow Patterns

Request input at decision points:

# Request approval before proceeding
case.request_human_input(
    CaseNodeUpdate(
        name="Approval Required",
        payload={
            "supervaizer_form": {
                "question": "Should we proceed with this operation?",
                "answer": {"fields": [...]}
            }
        },
        is_final=False,
    ),
    "Please review and approve before continuing."
)

# When human provides input, your custom receive_human_input agent method is called
# The handler function receives the form data in kwargs["fields"]
# Example handler:
# def handle_human_input(**kwargs) -> JobResponse:
#     fields = kwargs.get("fields", {})
#     if fields.get("Approved"):
#         # Continue with operation
#         case.update(CaseNodeUpdate(name="Approved, proceeding...", ...))
#     elif fields.get("Rejected"):
#         # Handle rejection
#         case.close(case_result={"status": "rejected"}, final_cost=0.0)

Use clear, actionable questions:

Be specific about what decision is needed
Provide context in the message parameter
Define form fields that match the decision type
Use appropriate field types (BooleanField for yes/no, CharField for text input)

What's next?

Return to the Controller Setup Guide to review agent configuration
Check the Model Reference for detailed field specifications
Explore API Reference for advanced usage
Review Case Management in Supervaize Studio for UI features

Overview​

Case Class and Methods​

Case.start()​

case.update()​

case.patch_step()​

case.request_human_input()​

Receiving Human Input ( human_answer Agent Method)​

case.close()​

CaseNodeUpdate​

Fields​

Usage Examples​

JobResponse​

Fields​

Usage Examples​

When to Use JobResponse vs Case Updates​

Manual Polling (job_poll)​

Scheduled Steps​

Scheduling a step​

How the executor works​

Workbench controls​

Cancellation on job stop​

REST API for scheduled steps​

Complete Example Flow​

Best Practices​

When to Create Multiple Cases vs Single Case​

Cost Tracking Patterns​

Error Handling Patterns​

Human Input Workflow Patterns​

What's next?​

Overview

Case Class and Methods

Case.start()

case.update()

case.patch_step()

case.request_human_input()

Receiving Human Input ( human_answer Agent Method)

case.close()

CaseNodeUpdate

Fields

Usage Examples

JobResponse

Fields

Usage Examples

When to Use JobResponse vs Case Updates

Manual Polling (job_poll)

Scheduled Steps

Scheduling a step

How the executor works

Workbench controls

Cancellation on job stop

REST API for scheduled steps

Complete Example Flow

Best Practices

When to Create Multiple Cases vs Single Case

Cost Tracking Patterns

Error Handling Patterns

Human Input Workflow Patterns

What's next?