step_confirmation.py
"""
Step Confirmation HITL Example (AgentOS)
This example demonstrates pausing a workflow before a step executes so the
human can approve or reject that step from AgentOS.
"""
from agno.os import AgentOS
from agno.workflow import OnReject
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 fetch_data(step_input: StepInput) -> StepOutput:
topic = step_input.input or "user data"
return StepOutput(
content=f"Fetched records for '{topic}'.\n"
"- 250 records found\n"
"- Sensitive fields detected\n"
"- Ready for processing"
)
def process_sensitive_data(step_input: StepInput) -> StepOutput:
previous_content = step_input.previous_step_content or "No fetched data"
return StepOutput(
content=f"Processed sensitive data:\n\n{previous_content}\n\n"
"- PII fields masked\n"
"- Aggregations calculated\n"
"- Processing audit created"
)
def save_results(step_input: StepInput) -> StepOutput:
previous_content = step_input.previous_step_content or "No processed data"
return StepOutput(
content=f"Saved workflow results.\n\nFinal payload:\n{previous_content}"
)
workflow = Workflow(
name="step_confirmation_workflow",
description="Pause before processing sensitive data and continue through AgentOS.",
db=db,
steps=[
Step(name="fetch_data", executor=fetch_data),
Step(
name="process_sensitive_data",
executor=process_sensitive_data,
requires_confirmation=True,
confirmation_message="Process sensitive data now?",
on_reject=OnReject.skip,
),
Step(name="save_results", executor=save_results),
],
)
workflow.id = "workflow-hitl-step-confirmation"
agent_os = AgentOS(
name="Workflow Step Confirmation HITL",
description="AgentOS workflow example for step-level confirmation.",
workflows=[workflow],
db=db,
)
app = agent_os.get_app()
if __name__ == "__main__":
agent_os.serve(app="step_confirmation:app", reload=True)
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]" psycopg-binary
3
Run PgVector
docker run -d \
-e POSTGRES_DB=ai \
-e POSTGRES_USER=ai \
-e POSTGRES_PASSWORD=ai \
-e PGDATA=/var/lib/postgresql \
-v pgvolume:/var/lib/postgresql \
-p 5532:5432 \
--name pgvector \
agnohq/pgvector:18
4
Run the example
Save the code blocks above as
step_confirmation.py and workflow_db.py in the same directory, then run:python step_confirmation.py