#!/usr/bin/env python3 """ CrewAI Human-in-the-Loop with AgentForms ========================================= This example shows a CrewAI agent that generates content but pauses to request human approval before publishing — a classic human-in-the-loop pattern. Pattern: 1. Agent generates a draft (e.g. blog post, report, email). 2. Agent uses the AgentForms SDK to create an approval form. 3. A shareable URL is sent to the human reviewer (Slack, email, etc.). 4. The agent blocks/polls until the human submits the form. 5. Based on the human's response, the agent continues or revises. Prerequisites: pip install crewai agentforms requests # Also configure your CrewAI LLM (e.g. OPENAI_API_KEY) """ import os import sys import time from typing import Optional # --------------------------------------------------------------------------- # AgentForms integration # --------------------------------------------------------------------------- try: from agentforms import AgentForms, FieldDefinition except ImportError: print("ERROR: agentforms SDK not installed. Run: pip install agentforms") sys.exit(1) class ApprovalGate: """ Creates an AgentForms approval form and blocks until a human submits it. This is the core human-in-the-loop primitive: the agent creates a form, shares the link, and polls the API for a response. """ def __init__(self, api_key: Optional[str] = None): self.client = AgentForms(api_key=api_key or os.environ["AGENTFORMS_API_KEY"]) def request_approval( self, title: str = "Action Approval Required", draft_content: str = "", context: str = "", timeout_minutes: int = 60, poll_interval: int = 15, ) -> dict: """ Create an approval form and wait for human response. Args: title: Shown as the form name. draft_content: The draft the human is asked to review. context: Extra context for the reviewer. timeout_minutes: Max wait time before raising TimeoutError. poll_interval: Seconds between API polls. Returns: dict with keys: "approved" (bool), "notes" (str), "revision_requested" (bool) """ # ---- Step 1: Build the approval form ---- 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": "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", "context": context}, ) 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"{'='*60}\n") # ---- Step 2: Poll for the human's submission ---- deadline = time.time() + (timeout_minutes * 60) while time.time() < deadline: result = self.client.submissions.list(form_token=form.token, limit=10) submissions = result.get("submissions", []) if submissions: # Take the most recent submission latest = submissions[-1] return self._parse_submission(latest) time.sleep(poll_interval) raise TimeoutError( f"No approval received within {timeout_minutes} minutes." ) def _parse_submission(self, submission) -> dict: """Extract approval decision from a Submission object.""" data = submission.dynamic_data approved_value = data.get("approved", "").strip() return { "approved": approved_value == "Yes", "revision_requested": approved_value == "No — revise", "stopped": approved_value == "No — stop", "notes": data.get("revision_notes", ""), } # --------------------------------------------------------------------------- # CrewAI agent definition # --------------------------------------------------------------------------- def build_crew_ai_agent() -> None: """ Full CrewAI example: agent drafts -> human approves -> agent acts. In a real setup you'd import Crew, Agent, Task from crewai. Below we simulate that flow so the file runs standalone for demo purposes. """ # ------------------------------------------------------------------ # 1. Initialise the approval gate # ------------------------------------------------------------------ gate = ApprovalGate() # ------------------------------------------------------------------ # 2. Simulate: CrewAI agent generates a draft # ------------------------------------------------------------------ # In production this would be: # # from crewai import Agent, Task, Crew, Process # # writer = Agent( # role="Content Writer", # goal="Write high-quality blog posts", # backstory="You are an experienced tech blogger.", # verbose=True, # ) # # draft_task = Task( # description="Write a 300-word blog post about 'The Future of AI Agents'", # agent=writer, # expected_output="A polished blog post ready for review.", # ) # # crew = Crew(agents=[writer], tasks=[draft_task], process=Process.sequential) # result = crew.kickoff() # draft = result.raw # # For the demo we use a placeholder draft. draft = ( "# The Future of AI Agents\n\n" "AI agents are rapidly evolving from simple chatbots to autonomous systems\n" "capable of planning, tool use, and human collaboration. This article\n" "explores where the industry is heading..." ) print("[CrewAI Agent] Draft generated. Requesting human approval...\n") # ------------------------------------------------------------------ # 3. Human-in-the-loop: request approval via AgentForms # ------------------------------------------------------------------ decision = gate.request_approval( title="Blog Post Approval", draft_content=draft, context="Draft written by CrewAI Content Writer agent.", timeout_minutes=30, ) # ------------------------------------------------------------------ # 4. Branch based on human decision # ------------------------------------------------------------------ if decision["approved"]: print("✅ Human approved. Publishing content.\n") # publish(draft) print(f"Published: {len(draft)} characters") elif decision["revision_requested"]: print(f"🔄 Revision requested.\n") print(f" Notes: {decision['notes']}\n") # Simulate revision loop revised_draft = ( draft + "\n\n[Revised section addressing reviewer feedback...]\n" ) print("[CrewAI Agent] Draft revised. Requesting re-approval...\n") decision2 = gate.request_approval( title="Blog Post — Revised Approval", draft_content=revised_draft, context="Revised based on: " + decision["notes"], ) if decision2["approved"]: print("✅ Revised draft approved. Publishing.\n") # publish(revised_draft) else: print("❌ Revised draft rejected. Work stopped.\n") elif decision["stopped"]: print("❌ Human stopped the workflow.\n") # Cleanup or archive the draft else: print("⚠️ Unexpected decision. Aborting.\n") # --------------------------------------------------------------------------- # Entry point # --------------------------------------------------------------------------- if __name__ == "__main__": if "AGENTFORMS_API_KEY" not in os.environ: print("Set the AGENTFORMS_API_KEY environment variable and try again.") print(" export AGENTFORMS_API_KEY='***'") sys.exit(1) build_crew_ai_agent()