#!/usr/bin/env python3
"""
AgentForms Webhook Receiver ā Human-in-the-Loop Pattern
========================================================
This example shows how to receive AgentForms submission webhooks and use
them to unblock a waiting AI agent. Instead of polling, the agent creates
a form with a webhook URL, then a FastAPI/Flask server receives the human's
response and signals the agent to continue.
Pattern:
1. Agent creates a form with a webhook URL pointing to this server.
2. Agent shares the form link with a human reviewer.
3. Human fills out the form; AgentForms POSTs to the webhook URL.
4. The webhook handler parses the payload and places the result
into a shared queue / channel that the agent is listening on.
5. The agent resumes execution with the human's decision.
Architecture:
āāāāāāāāāāāā share URL āāāāāāāāāāāāā webhook POST āāāāāāāāāāāāāāāā
ā Agent ā āāāāāāāāāāāāāāŗ ā Human ā āāāāāāāāāāāāāāāāŗ ā Webhook ā
ā (AI) ā āāāāāāāāāāāāāā ā (Browser)ā ā Receiver ā
āāāāāāāāāāāā resume āāāāāāāāāāāāā (async signal) āāāāāāāāāāāāāāāā
Prerequisites:
pip install fastapi uvicorn agentforms requests
# Or:
pip install flask agentforms requests
export AGENTFORMS_API_KEY='***'
python webhook_receiver.py
# The server starts on http://localhost:8000 (FastAPI) or
# http://localhost:5000 (Flask). Use ngrok for external access:
# ngrok http 8000
"""
import asyncio
import json
import os
import sys
import time
import uuid
from collections import defaultdict
from threading import Event
from typing import Any, Optional
# ---------------------------------------------------------------------------
# AgentForms SDK
# ---------------------------------------------------------------------------
try:
from agentforms import AgentForms, FieldDefinition, WebhookPayload
except ImportError:
print("ERROR: agentforms SDK not installed. Run: pip install agentforms")
sys.exit(1)
# ---------------------------------------------------------------------------
# Shared state: pending requests keyed by a unique session token
# ---------------------------------------------------------------------------
#
# In production you'd use Redis, Postgres, or a message queue.
# For this example we use in-memory dicts and threading Events.
#
class PendingRequests:
"""Thread-safe store for pending human-in-the-loop requests."""
def __init__(self):
# session_token -> {"event": threading.Event, "result": dict, "form_token": str}
self._pending: dict[str, dict] = {}
# All received submissions (audit log)
self._history: list[dict] = []
def register(self, session_token: str, form_token: str) -> Event:
"""Register a new pending request. Returns an Event to wait on."""
ev = Event()
self._pending[session_token] = {"event": ev, "result": None, "form_token": form_token}
return ev
def submit(self, payload: dict) -> Optional[str]:
"""
Process an incoming webhook payload. Returns the session token if
matched, or None if no matching pending request was found.
"""
self._history.append({"received_at": time.time(), "payload": payload})
site_id = payload.get("site_id")
submission_id = payload.get("submission_id")
# Match by site_id (form ID) ā look for any pending request for this form
for token, req in self._pending.items():
if req["form_token"] and True: # Simplified: match any pending request for this form
req["result"] = self._parse_payload(payload)
req["event"].set()
# Remove from pending after delivering
del self._pending[token]
return token
print(f" ā ļø Webhook received but no pending request matched.")
print(f" site_id={site_id}, submission_id={submission_id}")
return None
def wait(self, session_token: str, timeout: float = 300.0) -> Optional[dict]:
"""Block until the human responds or timeout expires."""
req = self._pending.get(session_token)
if not req:
return None
signaled = req["event"].wait(timeout=timeout)
if signaled:
return req["result"]
else:
# Timeout ā clean up
del self._pending[session_token]
return None
def _parse_payload(self, payload: dict) -> dict:
"""Extract human decision from webhook payload."""
fields = payload.get("fields", {})
approved_value = fields.get("approved", "").strip()
return {
"approved": approved_value == "Yes",
"revision_requested": approved_value == "No ā revise",
"stopped": approved_value == "No ā stop",
"notes": fields.get("revision_notes", ""),
"raw_fields": fields,
"submission_id": payload.get("submission_id"),
}
@property
def pending_count(self) -> int:
return len(self._pending)
@property
def history_count(self) -> int:
return len(self._history)
# Global store (shared between webhook handler and agent threads)
store = PendingRequests()
# ===========================================================================
# FASTAPI WEBHOOK RECEIVER
# ===========================================================================
def build_fastapi_app():
"""
Build a FastAPI app that receives AgentForms webhooks.
Run with:
from webhook_receiver import build_fastapi_app
import uvicorn
uvicorn.run(build_fastapi_app(), port=8000)
"""
try:
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
except ImportError:
print("FastAPI not installed. Run: pip install fastapi uvicorn")
return None
app = FastAPI(title="AgentForms Webhook Receiver", version="1.0.0")
@app.post("/webhook/agentforms")
async def webhook_handler(request: Request) -> JSONResponse:
"""Receive and process AgentForms submission webhooks."""
try:
body = await request.json()
except Exception:
return JSONResponse(status_code=400, content={"error": "Invalid JSON"})
print(f"\nšØ Webhook received:")
print(f" event: {body.get('event')}")
print(f" site_name: {body.get('site_name')}")
print(f" submission_id: {body.get('submission_id')}")
print(f" fields: {body.get('fields', {})}")
# Process: match to a pending request and signal the agent
session_token = store.submit(body)
if session_token:
print(f" ā
Delivered to pending session: {session_token}")
result = store._history[-1]["payload"]["fields"]
print(f" š Decision: approved={result.get('approved', 'N/A')}")
else:
print(f" ā¹ļø No matching pending request ā stored in audit log.")
return JSONResponse(content={"received": True})
@app.get("/health")
async def health():
return {
"status": "ok",
"pending_requests": store.pending_count,
"total_webhooks": store.history_count,
}
@app.get("/status")
async def status():
"""List all pending requests and recent webhook history."""
return {
"pending": list(store._pending.keys()),
"pending_count": store.pending_count,
"history_count": store.history_count,
"recent_submissions": [
{"submission_id": h["payload"].get("submission_id"),
"site_name": h["payload"].get("site_name"),
"fields": h["payload"].get("fields", {})}
for h in store._history[-10:]
],
}
return app
# ===========================================================================
# FLASK WEBHOOK RECEIVER (alternative)
# ===========================================================================
def build_flask_app():
"""
Build a Flask app that receives AgentForms webhooks.
Run with:
from webhook_receiver import build_flask_app
app = build_flask_app()
app.run(port=5000)
"""
try:
from flask import Flask, request, jsonify
except ImportError:
print("Flask not installed. Run: pip install flask")
return None
app = Flask(__name__)
@app.route("/webhook/agentforms", methods=["POST"])
def webhook_handler():
"""Receive and process AgentForms submission webhooks."""
body = request.get_json(force=True)
print(f"\nšØ Webhook received:")
print(f" event: {body.get('event')}")
print(f" site_name: {body.get('site_name')}")
print(f" submission_id: {body.get('submission_id')}")
print(f" fields: {body.get('fields', {})}")
session_token = store.submit(body)
if session_token:
print(f" ā
Delivered to pending session: {session_token}")
else:
print(f" ā¹ļø No matching pending request ā stored in audit log.")
return jsonify({"received": True}), 200
@app.route("/health")
def health():
return jsonify({
"status": "ok",
"pending_requests": store.pending_count,
"total_webhooks": store.history_count,
})
return app
# ===========================================================================
# AGENT SIDE: Create form + wait for webhook
# ===========================================================================
class WebhookApprovalGate:
"""
Agent-facing class that creates forms with webhook URLs and blocks
until the human responds (delivered via webhook to the receiver).
Usage:
gate = WebhookApprovalGate(webhook_url="https://your-server/webhook/agentforms")
decision = gate.request_approval("Deploy to prod?", draft_content=plan)
"""
def __init__(
self,
api_key: Optional[str] = None,
webhook_url: str = "http://localhost:8000/webhook/agentforms",
):
self.client = AgentForms(api_key=api_key or os.environ["AGENTFORMS_API_KEY"])
self.webhook_url = webhook_url
def request_approval(
self,
title: str = "Action Approval Required",
draft_content: str = "",
context: str = "",
timeout_minutes: int = 60,
) -> dict:
"""
Create an approval form with a webhook callback and wait for response.
The form is configured to POST submissions to self.webhook_url.
This method blocks until the webhook arrives (or times out).
Args:
title: Form title shown to the human.
draft_content: The draft being reviewed.
context: Extra context included in form metadata.
timeout_minutes: Max time to wait for human response.
Returns:
dict with keys: approved, revision_requested, stopped, notes, raw_fields
"""
# ---- Step 1: Generate a unique session token ----
session_token = f"session-{uuid.uuid4().hex[:12]}"
# ---- Step 2: Register the pending request ----
fields = [
{
"name": "approved",
"label": "Approve this action?",
"type": "select",
"options": ["Yes", "No ā revise", "No ā stop"],
"required": True,
},
{
"name": "revision_notes",
"label": "Feedback / revision notes",
"type": "textarea",
"required": False,
"placeholder": "What should be changed?",
},
{
"name": "session_token",
"label": "Session ID (do not change)",
"type": "hidden",
"default": session_token,
},
{
"name": "draft_content",
"label": "Draft (for reference)",
"type": "hidden",
"default": draft_content,
},
]
form = self.client.forms.create(
name=title,
fields=fields,
metadata={
"pattern": "human-in-the-loop",
"webhook_url": self.webhook_url,
"session_token": session_token,
"context": context,
},
)
# Register the pending request so the webhook handler can match it
ev = store.register(session_token, form.token)
share_url = form.share_url
print(f"\n{'='*60}")
print(f" š¤ HUMAN APPROVAL NEEDED")
print(f" š Open this link to approve:")
print(f" {share_url}")
print(f" š Webhook will POST to: {self.webhook_url}")
print(f" š Session: {session_token}")
print(f"{'='*60}\n")
# ---- Step 3: Wait for the webhook to arrive ----
decision = store.wait(
session_token,
timeout=timeout_minutes * 60,
)
if decision is None:
raise TimeoutError(
f"No approval received within {timeout_minutes} minutes "
f"(session: {session_token})"
)
return decision
# ===========================================================================
# FULL FLOW DEMO
# ===========================================================================
def demo_full_flow():
"""
Demonstrate the complete human-in-the-loop flow:
1. Start webhook server (in a thread)
2. Agent creates form + waits
3. Simulate webhook delivery (or use a real browser)
"""
print("=" * 60)
print(" AgentForms Webhook Receiver ā Full Flow Demo")
print("=" * 60)
print()
# ---- Try to start FastAPI server in background ----
server_started = False
server_process = None
try:
import subprocess
server_process = subprocess.Popen(
[sys.executable, "-c",
"from webhook_receiver import build_fastapi_app; "
"import uvicorn; "
"app = build_fastapi_app(); "
"uvicorn.run(app, host='127.0.0.1', port=8000, log_level='warning')"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
)
time.sleep(1.5) # Wait for server to start
server_started = True
print("ā
FastAPI webhook server started on http://localhost:8000")
except Exception as e:
print(f"ā ļø Could not start server: {e}")
print(" Run the server manually: uvicorn webhook_receiver:build_fastapi_app()")
# ---- Agent creates form and waits ----
gate = WebhookApprovalGate(
webhook_url="http://localhost:8000/webhook/agentforms",
)
draft = (
"## Deployment Plan\n"
"- Version: 2.4.1\n"
"- Changes: Bug fixes, performance improvements\n"
"- Rollback: Automated rollback configured\n"
)
try:
print("\nš¤ Agent: Created approval form. Waiting for human response...\n")
print(" (In a real scenario, a human would open the link above.)")
print(" For demo purposes, we'll simulate a webhook delivery.\n")
# Give the user time to check the form URL, then simulate webhook
print(" Simulating webhook delivery in 5 seconds...")
time.sleep(5)
# ---- Simulate webhook delivery ----
if server_started:
import requests
simulated_payload = {
"event": "submission",
"site_id": 42,
"site_name": "Deployment Approval",
"submission_id": 156,
"submitted_at": time.strftime("%Y-%m-%dT%H:%M:%SZ"),
"fields": {
"approved": "Yes",
"revision_notes": "Looks good, proceed with deployment.",
"session_token": "session-demo",
"draft_content": draft,
},
}
print("\nš¤ Simulating webhook POST...")
try:
resp = requests.post(
"http://localhost:8000/webhook/agentforms",
json=simulated_payload,
timeout=5,
)
print(f" Response: {resp.status_code} ā {resp.text}")
except requests.ConnectionError:
print(" ā ļø Could not connect to webhook server.")
print(" Make sure the server is running on port 8000.")
time.sleep(1)
except KeyboardInterrupt:
print("\n\nā¹ļø Demo interrupted.")
finally:
if server_process:
server_process.terminate()
server_process.wait(timeout=5)
print("\n" + "=" * 60)
print(" Demo complete!")
print("=" * 60)
# ===========================================================================
# ENTRY POINT
# ===========================================================================
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="AgentForms Webhook Receiver")
parser.add_argument(
"mode",
nargs="?",
default="server",
choices=["server", "flask", "demo"],
help="Run mode: 'server' (FastAPI), 'flask', or 'demo' (full flow)",
)
parser.add_argument("--port", type=int, default=8000, help="Server port")
args = parser.parse_args()
if args.mode == "server":
print("Starting FastAPI webhook receiver...")
try:
import uvicorn
app = build_fastapi_app()
if app:
uvicorn.run(app, host="127.0.0.1", port=args.port)
except ImportError:
print("uvicorn not installed. Run: pip install uvicorn")
elif args.mode == "flask":
print("Starting Flask webhook receiver...")
app = build_flask_app()
if app:
app.run(host="127.0.0.1", port=args.port, debug=True)
elif args.mode == "demo":
if "AGENTFORMS_API_KEY" not in os.environ:
print("Set the AGENTFORMS_API_KEY environment variable and try again.")
sys.exit(1)
demo_full_flow()