Confirm an ephemeral AgentOS tool call

Pause an agent run with @tool(requires_confirmation=True), inspect the pending tool returned by AgentOS, confirm it, and send the updated tool to the nested continue route.

Pause an agent run with @tool(requires_confirmation=True), inspect the pending tool returned by AgentOS, confirm it, and send the updated tool to the nested continue route. This pause exists only in the persisted run; it does not create an approval record.

basic.py

"""
Confirm an ephemeral AgentOS tool call
======================================

Pause an agent run with ``@tool(requires_confirmation=True)``, inspect the
pending tool returned by AgentOS, confirm it, and send the updated tool to the
nested ``continue`` route. This pause exists only in the persisted run; it does
not create an approval record.

Prerequisites: OPENAI_API_KEY
Run: .venvs/demo/bin/python cookbook/05_agent_os/05_human_in_the_loop/basic.py
Try: Run this file with --demo in another terminal
"""

import argparse
import json

import httpx
from agno.agent import Agent
from agno.db.sqlite import SqliteDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS
from agno.tools import tool

# ---------------------------------------------------------------------------
# Create Confirmation AgentOS
# ---------------------------------------------------------------------------

BASE_URL = "http://localhost:7777"
AGENT_ID = "confirmation-agent"
SESSION_ID = "confirmation-demo"


@tool(requires_confirmation=True)
def restart_service(service: str) -> str:
    """Restart one service after the caller confirms the action."""
    return f"Restarted {service}"


db = SqliteDb(
    id="confirmation-db",
    db_file="tmp/agent_os_hitl_basic.db",
)

confirmation_agent = Agent(
    id=AGENT_ID,
    name="Confirmation Agent",
    model=OpenAIResponses(id="gpt-5.5"),
    db=db,
    tools=[restart_service],
    instructions=(
        "When asked to restart a service, call restart_service immediately. "
        "Do not ask for confirmation in chat because the tool enforces it."
    ),
)

agent_os = AgentOS(
    id="confirmation-os",
    db=db,
    agents=[confirmation_agent],
)
app = agent_os.get_app()


def run_demo() -> None:
    """Pause, confirm the returned tool execution, and continue the same run."""
    with httpx.Client(base_url=BASE_URL, timeout=120.0) as client:
        response = client.post(
            f"/agents/{AGENT_ID}/runs",
            data={
                "message": "Restart the billing service.",
                "session_id": SESSION_ID,
                "stream": "false",
            },
        )
        response.raise_for_status()
        paused = response.json()
        if paused["status"] != "PAUSED":
            raise RuntimeError(f"Expected PAUSED, got {paused['status']}")

        pending_tools = paused.get("tools") or []
        if not pending_tools:
            raise RuntimeError("The paused run returned no pending tool")
        for pending_tool in pending_tools:
            if pending_tool.get("requires_confirmation"):
                pending_tool["confirmed"] = True

        continued_response = client.post(
            f"/agents/{AGENT_ID}/runs/{paused['run_id']}/continue",
            data={
                "tools": json.dumps(pending_tools),
                "session_id": paused["session_id"],
                "stream": "false",
            },
        )
        continued_response.raise_for_status()
        continued = continued_response.json()
        if continued["status"] != "COMPLETED":
            raise RuntimeError(f"Expected COMPLETED, got {continued['status']}")

        print(f"Initial status: {paused['status']}")
        print("Confirmed tool: restart_service")
        print(f"Final status: {continued['status']}")
        print(f"Result: {continued.get('content')}")


# ---------------------------------------------------------------------------
# Run Confirmation AgentOS
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--demo",
        action="store_true",
        help="Run the HTTP client against a server already listening on port 7777.",
    )
    args = parser.parse_args()

    if args.demo:
        run_demo()
    else:
        agent_os.serve(app=app, port=7777)

Run the Example

1. Set up your virtual environment

   Mac

   uv venv --python 3.12
   source .venv/bin/activate
   
2. Install dependencies

   uv pip install -U "agno[os]" openai
   
3. Export your OpenAI API key

   Mac/Linux

   export OPENAI_API_KEY="your_openai_api_key_here"
   
4. Run the example

   Save the code above as basic.py, then run:

   python basic.py
   

Full source: cookbook/05_agent_os/05_human_in_the_loop/basic.py