Framework guides

How to add human approvals to CrewAI

Bridge CrewAI review events into Contro1 for routing, approvals, and auditability.

CrewAI integrations work best when human review events carry execution IDs into Contro1 so the workflow can resume safely and only once.

Use the integration skill

Copy this skill link into your code agent to add CrewAI and Contro1 to your system.

Copy skill link

Key takeaways

CrewAI emits human-review webhooks; we turn each one into a Contro1 request and echo the decision back into the resume payload.
Task(on_failure=...) hooks let you promote task exceptions to human decisions instead of silent failures.
Idempotency uses crewai:{execution_id}:{task_id} so duplicate review events never create duplicate requests.
Use an agent-role goal line to teach the crew when a human reviewer is mandatory.

When to reach for Contro1 with CrewAI

CrewAI is a natural fit for long-running, multi-agent workflows where different tasks need different approvers. A crew of agents might handle research, drafting, and scheduling autonomously, but publishing or sending an email is exactly the kind of task that needs a human sign-off.

Our bridge subscribes to CrewAI's human-review events and produces one Contro1 request per task_id. The response feeds back into CrewAI through the standard resume payload, so nothing in CrewAI has to know Contro1 exists.

Installation

pip install

pip install centcom flask python-dotenv

.env

CENTCOM_API_KEY=your_centcom_api_key
CENTCOM_BASE_URL=https://api.contro1.com/api/centcom/v1
CENTCOM_WEBHOOK_SECRET=whsec_your_signing_secret

Basic integration

crewai_bridge.py

from fastapi import FastAPI, Request
import centcom

app = FastAPI()

@app.post("/crewai/review")
async def review(req: Request):
    event = await req.json()
    execution_id = event["execution_id"]
    task_id = event["task_id"]

    centcom.create_request(
        type="approval",
        question=event["review_prompt"],
        context=event["task_output"],
        required_role="reviewer",
        callback_url="https://ops.example.com/webhooks/crewai-resume",
        external_request_id=f"crewai:{execution_id}:{task_id}",
        metadata={"execution_id": execution_id, "task_id": task_id},
    )
    return {"status": "pending_human_review"}

@app.post("/webhooks/crewai-resume")
async def resume(req: Request):
    body = await req.json()
    approved = bool(body["response"].get("approved"))
    resume_payload = {
        "execution_id": body["metadata"]["execution_id"],
        "task_id": body["metadata"]["task_id"],
        "is_approve": approved,
        "human_feedback": body["response"].get("comment", "Approved" if approved else "Rejected"),
    }
    await crewai.resume_task(resume_payload)
    return {"status": "ok"}

Case continuity

Pass execution_id directly as correlation_id - no prefix or hashing needed. This groups every task review, callback mapping, and audit-only result for the same crew run into one case timeline.

Keep task_id in external_request_id because idempotency is per task review, not per execution.

crewai_case.py

created = client.create_protocol_request({
    "title": f"CrewAI review for {task_id}",
    "request_type": "review",
    "source": {"integration": "crewai", "run_id": execution_id, "workflow_id": task_id},
    "continuation": {"mode": "instruction", "callback_url": callback_url},
    "external_request_id": f"crewai:{execution_id}:{task_id}",
    "correlation_id": execution_id,
})

Audit records and cases reference

Logging autonomous actions

Use log_action when a CrewAI task completes allowed work without a human decision, or when your bridge maps operator feedback back into CrewAI.

Set in_reply_to when the audit record explains what happened after a specific Contro1 request.

crewai_audit.py

client.log_action(
    action="crewai.task_resume_mapped",
    summary=f"Mapped operator feedback to CrewAI task {task_id}",
    source={"integration": "crewai", "workflow_id": task_id, "run_id": execution_id},
    correlation_id=execution_id,
    in_reply_to={"type": "request", "id": created["id"]},
)

Gate the tool before it executes

The tool function itself is the right place to require approval for irreversible actions. The first line of a destructive tool calls Contro1 and blocks until an operator decides. Nothing runs until the human says yes - no prompt engineering needed.

tools/send_email.py

import os
import centcom
from crewai.tools import tool

_centcom = centcom.Client(api_key=os.environ["CENTCOM_API_KEY"])

@tool("send_email")
def send_email(to: str, subject: str, body: str) -> str:
    req = _centcom.create_request(
        type="approval",
        question=f"Send email to {to}: '{subject}'?",
        context={"to": to, "subject": subject},
        required_role="manager",
        external_request_id=f"email:{to}:{subject[:40]}",
    )
    decision = _centcom.wait_for_response(req["id"], timeout=600)
    if not (decision.get("response") or {}).get("approved"):
        raise PermissionError("Email rejected by operator")
    return mailer.send(to, subject, body)

Pause the agent on system error - orchestrator level

Attach an on_failure hook to risky Tasks. Instead of letting the crew fail silently and retry in a loop, promote the error to a human decision.

task_with_error_gate.py

from crewai import Task
import centcom

def on_task_failure(task, exc, execution_id):
    req = centcom.create_request(
        type="approval",
        question=f"Task {task.name} failed - retry?",
        context=f"{type(exc).__name__}: {exc}",
        required_role="oncall",
        priority="urgent",
        external_request_id=f"crewai-err:{execution_id}:{task.name}",
    )
    decision = centcom.wait_for_response(req["id"], timeout=900)
    if not decision["response"].get("approved"):
        raise RuntimeError("Stopped by human decision") from exc
    return task.execute()

publish_post = Task(
    description="Publish the drafted post to LinkedIn",
    agent=publisher_agent,
    on_failure=on_task_failure,
)

Escalate tool errors to a human

Inside tools your crew calls, escalate exceptions directly to Contro1 before they bubble back up. This keeps the operator close to the raw error context (stack trace, provider response) rather than CrewAI's abstraction of it.

tools/publish.py

import centcom
from crewai.tools import tool

@tool("publish_post")
def publish_post(draft_id: str) -> str:
    try:
        return cms.publish(draft_id)
    except cms.RateLimitError as exc:
        req = centcom.create_request(
            type="approval",
            question=f"CMS rate-limited on draft {draft_id}. Retry in 5 min?",
            context=str(exc),
            required_role="content-ops",
            external_request_id=f"publish-err:{draft_id}",
        )
        decision = centcom.wait_for_response(req["id"], timeout=900)
        if decision["response"].get("approved"):
            return cms.publish(draft_id)
        return "skipped by operator"

Prompt engineering: force the agent to pause

Use the Agent role/goal/backstory to instruct the crew member that certain categories of output must always go through human review.

agents.yaml

publisher:
  role: >
    Content publisher for Marketing
  goal: >
    Publish approved drafts to LinkedIn and the blog. You never publish without
    a human reviewer approving the final copy via CENTCOM. If CENTCOM is
    unreachable, stop and report. Never publish speculatively.
  backstory: >
    You are cautious by policy: irreversible public actions always require a
    human sign-off, even when the draft looks perfect.
  constraints:
    - Before calling publish_post, write the draft to review_drafts and emit a
      human-review event with review_prompt="Publish this draft to LinkedIn?".
    - On rejection, do not retry with a rewritten draft. Return the reviewer's
      comment to the requesting agent.

See our GitHub integration repo

Reference implementation for the CrewAI bridge, including resume payload mapping and deduplication guidance.

centcom-crewai on GitHub · crewai_bridge.py - full bridge example

Frequently asked questions

Why is correlation metadata so important in CrewAI?

Because you need to resume the exact execution and task safely after asynchronous human review. Pass execution_id and task_id through metadata on every request.

How do I handle duplicate review events?

Use crewai:{execution_id}:{task_id} as external_request_id. Contro1 returns the existing request instead of creating a second one.

Can I enforce approval at the crew level instead of per task?

Yes. Gate the final crew output with a single review task whose job is to emit the CrewAI human-review event before anything ships.

What if the reviewer rejects - does the crew retry forever?

No. Our bridge sets is_approve=false and passes the reviewer's comment into human_feedback. The crew sees this as a terminal "do not publish" signal unless your flow explicitly redirects to a redraft task.