How to add human approvals to CrewAI

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 centcom flask python-dotenv

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

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"}

Send context the reviewer can trust

A request that only says "Approve this tool call?" forces the reviewer to rubber-stamp. Send three things with every gated call: the exact tool input your gate intercepted (machine-observed fact), the agent's own justification (make reason a required parameter of the risky tool, so the model produces it at decision time), and the trigger - the user message or event that started the run.

Keep the two kinds apart in context: facts your code observed versus text the model wrote. Agent-written justification is agent-reported evidence: it helps the reviewer decide, but it must never change routing, risk_level, or approval policy, because a prompt-injected agent writes very persuasive reasons. If a high-risk request arrives without this context, fail closed and reject it instead of asking a human to guess.

Reviewer context and trust rules

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.

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.

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.

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.

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.

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.

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

Send full agent traceability

Beyond the approval call, attach identity, a run trace, the tools you invoked, and the context you retrieved. Each field is optional — add what you have. The verified identity always comes from your API key; a caller-supplied actor.agent_id is recorded as a claimed sub-agent until an admin verifies it.

• trace_id / parent_trace_id — link one run (and sub-agent runs) into a single trace.
• tool_calls[] — what the agent tried to do, so reviewers see the actions.
• retrieved_context[] — the data the decision was based on (RAG provenance).
• Then export a signed evidence packet from GET /requests/:id/evidence.

# Same approval call you already make — now with full traceability.
client.requests.create(
    request_type="approval",
    title="Refund $4,200 to customer 8831",
    source={"integration": "crewai"},
    actor={"agent_id": "billing-agent", "agent_name": "Billing Agent"},  # claimed sub-agent
    trace_id=f"trc_{run_id}",          # link every step of this run
    tool_calls=[{"name": "lookup_order", "outcome": "success"}],
    retrieved_context=[{"source": "policy:refunds", "uri": "kb://policy/refunds"}],
    continuation={"mode": "decision"},
)

Agent identity, traceability & signed evidence

Make the approval the gate, not a suggestion

The signed webhook is cryptographic proof of a human decision. Verify it inside the system that executes the action - not inside the agent. Any tool that must never run without human sign-off (payments, deploys, data deletion) should refuse to act without a verified approval; that way no agent, including shadow agents nobody registered, can trigger it by skipping Contro1.

• Verify the signature and reject timestamps older than 5 minutes (replay protection; the timestamp marks callback delivery, not request creation, so long SLAs are unaffected).
• Bind the approval to the exact action parameters via metadata / correlation_id - never treat "an approval arrived" as permission for a different action.
• Execute each request_id exactly once (idempotency on your side).
• When in doubt, confirm state directly with GET /v1/requests/:id using a read-only API key.

Full guardrail pattern with code: Webhooks

Put this integration in the enterprise control model

The connector creates the runtime gate. The operating model around it defines which actions pause, which owner receives the decision, how a timeout escalates, and where the outcome is recorded across agent frameworks.

runtime human approval for AI agents · enterprise AI agent control plane

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.

Related resources