team_approval_cli.py

"""
Team Approval: member agent tool with @approval decorator.

Demonstrates the approval system on a team — the tool pauses for both
HITL confirmation AND creates an approval record in the DB that an admin
can review.

Based on cookbook/02_agents/11_approvals/approval_team.py
"""

import time

from agno.agent import Agent
from agno.approval import approval
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.team.team import Team
from agno.tools import tool

DB_URL = "postgresql+psycopg://ai:ai@localhost:5532/ai"

@approval
@tool(requires_confirmation=True)
def deploy_to_production(app_name: str, version: str) -> str:
    """Deploy an application to production.

    Args:
        app_name (str): Name of the application.
        version (str): Version to deploy.
    """
    return f"Successfully deployed {app_name} v{version} to production"


if __name__ == "__main__":
    # Clean slate
    db = PostgresDb(db_url=DB_URL, session_table="team_sessions", approvals_table="approvals")

    deploy_agent = Agent(
        id="deploy-agent",
        name="Deploy Agent",
        role="Handles deployments to production",
        model=OpenAIResponses(id="gpt-5-mini"),
        tools=[deploy_to_production],
    )

    team = Team(
        id="devops-team",
        name="DevOps Team",
        members=[deploy_agent],
        model=OpenAIResponses(id="gpt-5-mini"),
        db=db,
    )

    # Step 1: Run — should pause for confirmation
    print("--- Step 1: Run team (should pause) ---")
    response = team.run("Deploy the payments app version 2.1 to production")
    print(f"  Status: {response.status}")

    if not response.is_paused:
        print(f"  FAIL: Not paused. Content: {str(response.content)[:100]}")
        exit(1)

    print("  PASS: Team paused")

    # Step 2: Check approval record exists in DB
    print("\n--- Step 2: Check approval record ---")
    approvals_list, total = db.get_approvals(status="pending")
    print(f"  Pending approvals: {total}")

    if total >= 1:
        record = approvals_list[0]
        print(f"  Approval ID: {record['id']}")
        print(f"  Source type: {record['source_type']}")
        print(f"  Source name: {record.get('source_name')}")
    else:
        print("  WARN: No approval record (model may not have delegated)")

    # Step 3: Confirm requirements and continue
    print("\n--- Step 3: Confirm and continue ---")
    for req in response.active_requirements:
        if req.needs_confirmation:
            print(f"  Confirming: {req.tool_execution.tool_name}({req.tool_execution.tool_args})")
            req.confirm()

    result = team.continue_run(response)
    print(f"  Status after continue: {result.status}")
    print(f"  Content: {str(result.content)[:150]}")

    # Step 4: Resolve approval in DB
    print("\n--- Step 4: Resolve approval in DB ---")
    if total >= 1:
        resolved = db.update_approval(
            record["id"],
            expected_status="pending",
            status="approved",
            resolved_by="devops_lead",
            resolved_at=int(time.time()),
        )
        if resolved:
            print(f"  Resolved: {resolved['status']} by {resolved['resolved_by']}")
        else:
            print("  WARN: Could not resolve (may already be resolved)")

    # Step 5: Verify no pending approvals
    print("\n--- Step 5: Verify ---")
    remaining = db.get_pending_approval_count()
    print(f"  Pending approvals remaining: {remaining}")

    print("\n--- Done ---")