Configuring ApprovalHandler Workflows with WebhookApproval for Human-in-the-Loop in AGT

Agent Governance Toolkit (AGT) enables human-in-the-loop controls by routing approval requests through webhook endpoints using the WebhookApproval handler, which POSTs structured JSON payloads to external services and awaits explicit human decisions before allowing sensitive agent actions to proceed.

Agent Governance Toolkit (AGT) provides a policy-driven framework for enforcing human oversight on autonomous agent operations. When you configure ApprovalHandler workflows with WebhookApproval, you create a secure gate that pauses execution until a human explicitly validates high-risk actions through external systems like Slack, Microsoft Teams, or custom approval portals.

Understanding the Approval Handler Architecture

AGT implements four distinct approval adapters, each suited for different environments:

• CallbackApproval executes a user-supplied Python function for synchronous, in-process decisions ideal for unit tests.
• ConsoleApproval reads input from STDIN, suitable for local development and debugging.
• WebhookApproval dispatches HTTP POST requests to external endpoints, enabling production integrations with ticketing systems, chat platforms, or dedicated approval services.
• AutoRejectApproval immediately denies all requests, serving as a fail-safe when no handler is configured.

When a policy rule containing action: require_approval matches, the governance engine instantiates an ApprovalRequest (defined in agent-governance-python/agent-mesh/src/agentmesh/governance/approval.py, lines 33-55) and passes it to your configured handler. For webhook-based workflows, this triggers an HTTP round-trip that blocks the agent until a response is received or a timeout occurs.

Webhook Message Protocol

Request Payload Structure

The WebhookApproval handler dispatches a POST request containing a JSON document with contextual metadata about the pending operation:

{
  "type": "approval_request",
  "rule_name": "approve-large-transfer",
  "policy_name": "financial-approvals",
  "agent_id": "*",
  "action": "transfer",
  "approvers": ["treasury-ops", "compliance"],
  "requested_at": "2026-04-23T12:00:00Z"
}

Required Response Format

Your endpoint must return HTTP 200 with a JSON body containing three mandatory fields:

{
  "approved": true,
  "approver": "[email protected]",
  "reason": "Reviewed and approved"
}

If the webhook returns malformed data, hits a network timeout, or responds with a non-200 status code, the handler defaults to denial. As implemented in agent-governance-python/agent-mesh/src/agentmesh/governance/approval.py (lines 250-257), WebhookApproval catches all request exceptions and returns an ApprovalDecision with approved: false, ensuring that communication failures cannot inadvertently authorize restricted actions.

Security and Validation

Before dispatching any request, AGT sanitizes the target URL using _validate_webhook_url (located in agent-governance-python/agent-mesh/src/agentmesh/governance/advisory.py, lines 45-58). This validation rejects dangerous URL schemes and known SSRF-prone hosts, preventing attackers from using the approval mechanism to probe internal networks or access cloud metadata services.

Step-by-Step Implementation

1. Define the Policy Gate

Create a YAML policy that triggers approval requirements for specific actions. The following example from docs/tutorials/policy-as-code/examples/05_approval_policy.yaml mandates human review for all financial transfers:

# financial-approval-policy.yaml

apiVersion: governance.toolkit/v1
name: financial-approvals
agents: ["*"]
default_action: allow

rules:
  - name: approve-large-transfer
    condition: "action.type == 'transfer'"
    action: require_approval
    approvers: ["treasury-ops", "compliance"]
    description: "All transfers require human approval"
    priority: 100

2. Wire the WebhookApproval Handler

Instantiate the handler with your endpoint, timeout duration, and authentication headers. This pattern appears in the official tutorial at docs/tutorials/38-approval-workflows.md (lines 102-115):

from agentmesh.governance import govern, WebhookApproval

def transfer_funds(action, amount, to_account):
    print(f"💰 Transferring ${amount} → {to_account}")
    return {"transferred": True, "amount": amount}

approval_handler = WebhookApproval(
    url="https://hooks.example.com/approval",
    timeout_seconds=300,
    headers={"Authorization": "Bearer <SLACK_TOKEN>"}
)

safe_transfer = govern(
    transfer_funds,
    policy="financial-approval-policy.yaml",
    approval_handler=approval_handler,
)

result = safe_transfer(action="transfer", amount=12_000, to_account="ACC-123")

3. Implement the Webhook Receiver

Build an HTTP service that receives the AGT payload and presents it to human reviewers. Below is a minimal Flask implementation that auto-approves small transfers while escalating larger amounts:

from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route("/approval", methods=["POST"])
def approval():
    data = request.json
    if data["action"] == "transfer" and data.get("amount", 0) < 5000:
        return jsonify(
            approved=True,
            approver="auto-service",
            reason="Auto-approved small transfer"
        )
    return jsonify(
        approved=False,
        approver="pending-human",
        reason="Requires manual review"
    )

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8000)

4. Inspect the Audit Trail

Every decision is recorded immutably. Query the audit log to maintain compliance records:

for entry in safe_transfer.audit_log.query(event_type="approval_decision"):
    print(f"{entry.action} → {entry.outcome}")
    print(f"  Approver: {entry.data['approver']}")
    print(f"  Reason:   {entry.data['reason']}")

Summary

• WebhookApproval POSTs structured JSON to external endpoints and blocks agent execution until receiving an explicit approved: true response.
• URL validation via _validate_webhook_url prevents SSRF attacks by filtering dangerous schemes and internal hosts.
• The response must include approved, approver, and reason fields; any network or parsing error defaults to denial.
• Implementation requires a YAML policy with action: require_approval, a WebhookApproval instance configured with timeouts and headers, and an external HTTP service to route requests to humans.
• All decisions generate tamper-evident audit entries, creating compliance-ready trails of who authorized specific agent actions.

Frequently Asked Questions

What happens if the webhook endpoint is unreachable or returns a 500 error?

If the HTTP request fails, times out, or returns malformed JSON, WebhookApproval returns a denied decision by default. According to the source code in agent-governance-python/agent-mesh/src/agentmesh/governance/approval.py (lines 250-257), the handler catches all exceptions during the request cycle and treats them as rejection events, ensuring that network instability cannot bypass governance controls.

Can I integrate WebhookApproval with Slack or Microsoft Teams?

Yes. Configure the url parameter to point to your middleware service (such as a Slack webhook URL or Teams connector), and include appropriate authentication headers like Authorization: Bearer <token>. Your service translates the AGT payload into platform-specific interactive messages, captures user responses, and returns the standardized JSON decision that AGT requires.

How do I test approval workflows without external HTTP calls?

Use the CallbackApproval handler for unit tests. As demonstrated in agent-governance-python/agent-os/tests/test_human_approval_all_adapters.py, inject a synchronous Python function that immediately returns an ApprovalDecision. This allows you to test policy logic without network dependencies, reserving WebhookApproval for integration and production environments.

Is there a maximum timeout limit for webhook approvals?

The timeout_seconds parameter accepts any positive integer, though practical limits depend on your HTTP client configuration and the patience of the calling agent process. Values exceeding 300 seconds (5 minutes) may require adjusting underlying connection pool settings or session timeouts in your AGT deployment to prevent premature client-side aborts.

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