Human-in-the-Loop Example

AgentOS with tools requiring user confirmation

This example shows how to implement Human-in-the-Loop (HITL) flows in AgentOS. When an agent needs to execute a tool that requires confirmation, the run pauses and waits for user approval before proceeding.

Prerequisites

• Python 3.10 or higher
• PostgreSQL with pgvector (setup instructions below)
• OpenAI API key

Code

hitl_confirmation.py

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.tools import tool

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


@tool(requires_confirmation=True)
def delete_records(table_name: str, count: int) -> str:
    """Delete records from a database table.

    Args:
        table_name: Name of the table
        count: Number of records to delete

    Returns:
        str: Confirmation message
    """
    return f"Deleted {count} records from {table_name}"


@tool(requires_confirmation=True)
def send_notification(recipient: str, message: str) -> str:
    """Send a notification to a user.

    Args:
        recipient: Email or username of the recipient
        message: Notification message

    Returns:
        str: Confirmation message
    """
    return f"Sent notification to {recipient}: {message}"


# Create agent with HITL tools
agent = Agent(
    name="Data Manager",
    id="data_manager",
    model=OpenAIResponses(id="gpt-5.2"),
    tools=[delete_records, send_notification],
    instructions=["You help users manage data operations"],
    db=db,
    markdown=True,
)

# Create AgentOS
agent_os = AgentOS(
    id="agentos-hitl",
    agents=[agent],
)

app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="hitl_confirmation:app", port=7777)

Testing the Example

Once the server is running, test the HITL flow:

# 1. Send a request that requires confirmation
curl -X POST http://localhost:7777/agents/data_manager/runs \
  -F "message=Delete 50 old records from the users table" \
  -F "user_id=test_user" \
  -F "session_id=test_session"

# The response will have status: "paused" with tools awaiting confirmation

# 2. Continue with approval (use the run_id and tool_call_id from response)
curl -X POST http://localhost:7777/agents/data_manager/runs/{run_id}/continue \
  -F 'tools=[{"tool_call_id":"{tool_call_id}","tool_name":"delete_records","tool_args":{"table_name":"users","count":50},"confirmed":true}]' \
  -F "session_id=test_session" \
  -F "user_id=test_user" \
  -F "stream=false"

# 3. Continue after admin approval is resolved in DB
curl -X POST http://localhost:7777/agents/data_manager/runs/{run_id}/continue \
  -F "tools=" \
  -F "session_id=test_session" \
  -F "user_id=test_user" \
  -F "stream=false"

If an admin resolves a required approval in the database, you can continue with an empty tools value. Non-admin users receive 403 if a required approval is still pending.

Approval Status API

Poll a single approval record:

GET /approvals/{approval_id}/status

When a run pauses, each tool that requires approval is stamped with approval_id. Store that ID and poll for status changes. Admins with the approvals:write scope can continue once the approval is resolved.

Usage

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   
2. Set Environment Variables

   export OPENAI_API_KEY=your_openai_api_key
   
3. Install dependencies

   uv pip install -U agno fastapi uvicorn sqlalchemy pgvector psycopg openai
   
4. Setup PostgreSQL Database

   docker run -d \
     --name agno-postgres \
     -e POSTGRES_DB=ai \
     -e POSTGRES_USER=ai \
     -e POSTGRES_PASSWORD=ai \
     -p 5532:5432 \
     pgvector/pgvector:pg17
   
5. Run Example

   python hitl_confirmation.py
   

Learn More

HITL Getting Started

Learn HITL basics with a simple example

HITL Concepts

Understand HITL patterns and best practices