🍳 Cookbook: Human-in-the-Loop Workflows

For sensitive actions—like sending emails, executing refunds, or deploying code—you should never let an agent run fully autonomously. This guide shows how to implement an Approval Gate.

🎯 Goal

Build a workflow that prepares a draft email but pauses for human approval before sending it.

---

🏗️ Phase 1: Define the Workflow

We use the SpecWorkflow logic but manually construct the Task list to include a dependency break.

# app/services/email_workflow.py
from app.models.thread import Task, ThreadStatus
from app.services.thread_manager import ThreadManager

async def start_email_draft(topic: str, recipient: str):
    tm = ThreadManager()

    # Define tasks
    tasks = [
        # Task 1: Auto-execute
        Task(
            id="draft_email", 
            agent="writer",
            input=f"Write a polite email about {topic} to {recipient}"
        ),

        # Task 2: This task depends on 'draft_email'
        # Crucially, we will configuring the executor to PAUSE here.
        Task(
            id="send_email",
            agent="executor", # Assuming executor has the send_email tool
            input="Send the email drafted in the previous step.",
            dependencies=["draft_email"],
            requires_human_approval=True  # 👈 The magic flag
        )
    ]

    thread = await tm.create(goal="Send email", tasks=tasks)
    return thread

---

⏸️ Phase 2: The Approval Trap

In your execution loop, you check for the flag.

# app/services/run_loop.py

async def execute_thread(thread_id):
    thread = get_thread(thread_id)

    for task in thread.tasks:
        if task.status == "pending":

            # CHECK FOR APPROVAL
            if task.requires_human_approval and not task.approved:
                print(f"⏸️  Task {task.id} requires approval. Pausing thread.")
                thread.status = ThreadStatus.PAUSED
                save_thread(thread)
                return  # <--- STOP EXECUTION HERE

            # Otherwise, execute...
            execute_task(task)

---

🚦 Phase 3: The Human Interface

The thread is now saved in the database in a PAUSED state. You can build a simple CLI or UI to list and approve tasks.

1. List Pending Approvals

poetry run agentic approvals list
# Found 1 paused thread:
# Thread ID: th_123
# Pending Task: "send_email"
# Context: "Subject: Hello..." (Draft content)

2. Approve Execution

This command flips the boolean flag and resumes the loop.

poetry run agentic approvals approve th_123 task_send_email
# ✅ Task approved. Resuming thread execution...
# 📧 Email sent!

---

🛡️ Implementation in Traylinx Template

The AsyncExecutor in the Traylinx template has native support for this.

1. Mark the Task: When creating a Task model, set require_approval=True.
2. Run: The executor automatically halts when it hits that task.
3. Resume: Call executor.resume_thread(thread_id, approval_data={...}).

This pattern ensures that no critical action happens without an explicit human "Yes".