How to Implement Human-in-the-Loop Approval Workflows in Agno

Agno provides a first-class @approval decorator that marks any @tool as requiring human review, automatically pausing execution via runtime helpers in libs/agno/agno/run/approval.py until a human approves, rejects, or provides input through the persistence layer.

Human-in-the-loop (HITL) approval workflows in the agno-agi/agno repository allow agents to pause execution when encountering critical operations, requiring explicit human confirmation before proceeding. By combining the @approval decorator with the built-in runtime gating logic, you can implement both blocking approvals and non-blocking audit trails with minimal boilerplate. This article explains how to implement human-in-the-loop approval workflows in Agno using the actual source code architecture.

Core Architecture Components

The @approval Decorator

Located in libs/agno/agno/approval/decorator.py, the @approval decorator attaches metadata to any function decorated with @tool. It sets func.approval_type to either "required" or "audit" and validates HITL flags such as requires_confirmation, requires_user_input, or external_execution_required.

Approval Types and Persistence

The ApprovalType enum in libs/agno/agno/approval/types.py defines two modes:

• required: Blocks execution until human resolution
• audit: Logs the interaction without blocking

The Approval dataclass in libs/agno/agno/db/schemas/approval.py defines the database schema, storing fields like run_id, tool_name, status (pending, approved, rejected), and resolution_data.

Runtime Execution Gating

The libs/agno/agno/run/approval.py module contains the orchestration logic. When a run encounters an approved tool, _has_approval_requirement detects the requirement, triggering create_approval_from_pause to persist a pending record via _build_approval_dict. Upon resumption, check_and_apply_approval_resolution (or its async variant acheck_and_apply_approval_resolution) gates execution based on the stored status.

Implementing Blocking Approvals

Use the @approval decorator without arguments (or with type="required") to force the agent to pause until a human confirms the action.

from agno.approval import approval
from agno.tools import tool

@approval
@tool(requires_confirmation=True)
def delete_production_database(db_name: str) -> str:
    """Delete a production database - requires explicit human approval."""
    # Implementation here

    return f"Database {db_name} deleted"

When the agent invokes this tool, the runtime calls create_approval_from_pause to store a record with pause_type="confirmation" and status pending. The workflow cannot proceed until external code updates the record to approved or rejected.

Implementing Audit-Only Workflows

For operations that should complete immediately but leave an auditable trail, use type="audit".

from agno.approval import approval, ApprovalType
from agno.tools import tool

@approval(type=ApprovalType.AUDIT)
@tool()
def export_user_data(user_id: str) -> str:
    """Export sensitive user data - logged for compliance."""
    # Export logic here

    return f"Data exported for user {user_id}"

In this mode, create_audit_approval generates the record after execution completes, capturing the outcome without blocking the run.

Runtime Flow and Resolution

When a workflow encounters a tool marked for approval:

1. Pause Detection: The runner checks _has_approval_requirement and identifies the approval_type.
2. Record Creation: create_approval_from_pause builds an approval dict via _build_approval_dict and persists it via db.create_approval, storing metadata including run_id, tool_name, and context.
3. Human Resolution: An external UI or API updates the database record, setting status to approved or rejected and optionally populating resolution_data with user inputs.
4. Gating Logic: Before the next step, check_and_apply_approval_resolution calls _get_approval_for_run to fetch the latest record.
   • If pending: Raises RuntimeError to halt execution.
   • If approved: _apply_approval_to_tools injects confirmation flags or user data into the Function object.
   • If rejected: Marks the tool as confirmed=False, allowing error handling logic to execute.

Complete Workflow Example

Here is a complete example demonstrating how to run a workflow with mixed approval types:

from agno.run import run_workflow
from agno.approval import approval
from agno.tools import tool

@approval
@tool(requires_confirmation=True)
def deploy_to_production(commit_hash: str) -> str:
    """Deploy code to production servers."""
    return f"Deployed {commit_hash}"

@approval(type="audit")
@tool()
def notify_team(channel: str) -> str:
    """Send notification to team channel."""
    return f"Notification sent to {channel}"

workflow = [deploy_to_production, notify_team]
run_id = run_workflow(workflow, agent_id="deploy-agent-001")

In this workflow, deploy_to_production triggers a pause via create_approval_from_pause until manually approved, while notify_team executes immediately and creates an audit record via the post-execution audit mechanism.

Summary

• Decorate tools with @approval from libs/agno/agno/approval/decorator.py to enable HITL workflows.
• Use type="required" (default) for blocking approvals that pause execution until human resolution.
• Use type="audit" from libs/agno/agno/approval/types.py for non-blocking compliance logging.
• Runtime gating occurs in libs/agno/agno/run/approval.py via check_and_apply_approval_resolution, which enforces pending status checks before allowing continuation.
• Approval records follow the Approval schema in libs/agno/agno/db/schemas/approval.py, storing run_id, tool_name, resolution status, and user input data.

Frequently Asked Questions

What happens if a required approval is rejected?

When an approval record is marked rejected, the _apply_approval_to_tools function sets confirmed=False on the tool's Function object. The agent runtime can then catch this state and skip execution or trigger error handling logic without performing the protected operation.

Can I collect user input during the approval process?

Yes. When defining the tool, set requires_user_input=True. During resolution, the human reviewer can provide resolution_data containing input values. The runtime injects these values into the tool via _apply_approval_to_tools before the agent continues execution.

How do I check the status of a pending approval programmatically?

Query the persistence layer using the run_id and tool identifier. The runtime uses _get_approval_for_run in libs/agno/agno/run/approval.py to fetch the latest approval record and verify whether the status has moved from pending to approved or rejected.

Does the decorator order matter when combining @approval and @tool?

No, the decorators are order-independent. The @approval decorator attaches metadata that @tool later reads, regardless of stacking order. The unit tests in libs/agno/tests/unit/tools/test_approval_decorator.py verify this behavior across different decoration sequences.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}