Skip to main content
A human-controlled Condition. Confirming runs the primary branch; rejecting runs the else branch.
condition_user_decision.py
"""
Condition User Decision HITL Example (AgentOS)

This example demonstrates a human-controlled Condition. Confirming runs the
primary branch; rejecting runs the else branch.
"""

from agno.os import AgentOS
from agno.workflow import OnReject
from agno.workflow.condition import Condition
from agno.workflow.step import Step
from agno.workflow.types import StepInput, StepOutput
from agno.workflow.workflow import Workflow
from workflow_db import db


def analyze_data(step_input: StepInput) -> StepOutput:
    topic = step_input.input or "Q4 sales data"
    return StepOutput(
        content=f"Initial analysis complete for '{topic}'.\n"
        "- Detailed review is available\n"
        "- Quick summary is also available"
    )


def detailed_analysis(step_input: StepInput) -> StepOutput:
    return StepOutput(
        content="Detailed analysis results:\n"
        "- Full statistical review completed\n"
        "- Edge cases examined\n"
        "- Processing time estimate: 10 minutes"
    )


def quick_summary(step_input: StepInput) -> StepOutput:
    return StepOutput(
        content="Quick summary results:\n"
        "- Key metrics computed\n"
        "- Top highlights identified\n"
        "- Processing time estimate: 1 minute"
    )


def generate_report(step_input: StepInput) -> StepOutput:
    previous_content = step_input.previous_step_content or "No analysis output"
    return StepOutput(content=f"Generated report:\n\n{previous_content}")


workflow = Workflow(
    name="condition_user_decision_workflow",
    description="Use AgentOS confirmation to choose a Condition branch.",
    db=db,
    steps=[
        Step(name="analyze_data", executor=analyze_data),
        Condition(
            name="analysis_depth_decision",
            requires_confirmation=True,
            confirmation_message="Run detailed analysis? Reject to use quick summary.",
            on_reject=OnReject.else_branch,
            steps=[Step(name="detailed_analysis", executor=detailed_analysis)],
            else_steps=[Step(name="quick_summary", executor=quick_summary)],
        ),
        Step(name="generate_report", executor=generate_report),
    ],
)
workflow.id = "workflow-hitl-condition-user-decision"

agent_os = AgentOS(
    name="Workflow Condition Decision HITL",
    description="AgentOS workflow example for human-controlled conditions.",
    workflows=[workflow],
    db=db,
)
app = agent_os.get_app()


if __name__ == "__main__":
    agent_os.serve(app="condition_user_decision:app", reload=True)
The example imports this helper module from the same directory:
workflow_db.py
from agno.db.postgres import PostgresDb

db_url = "postgresql+psycopg://ai:ai@localhost:5532/ai"
db = PostgresDb(db_url=db_url)

​
Run the Example

1

Set up your virtual environment

uv venv --python 3.12
source .venv/bin/activate

uv venv --python 3.12
.venv\Scripts\activate
2

Install dependencies

uv pip install -U "agno[os]" cel-python fastmcp psycopg-binary starlette
3

Export your API keys

export JWT_VERIFICATION_KEY="your_jwt_verification_key_here"

$Env:JWT_VERIFICATION_KEY="your_jwt_verification_key_here"
4

Run PgVector

docker run -d \
  -e POSTGRES_DB=ai \
  -e POSTGRES_USER=ai \
  -e POSTGRES_PASSWORD=ai \
  -e PGDATA=/var/lib/postgresql/data/pgdata \
  -v pgvolume:/var/lib/postgresql/data \
  -p 5532:5432 \
  --name pgvector \
  agnohq/pgvector:18
5

Run the example

Save the code blocks above as condition_user_decision.py and workflow_db.py in the same directory, then run:
python condition_user_decision.py
Full source: cookbook/05_agent_os/human_in_the_loop/workflow/condition_user_decision.py