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.

Connect your CrewAI agent to Contro1

A CrewAI agent runs in your own code, so it connects with an Agent Credential: a key bound to one agent, so every call is attributed to it and nothing in a request can change which agent it is.

• 1. Register the agent: contro1 init --name "<name>" --framework crewai, and finish the setup link it prints (purpose and owner).
• 2. Connect the application account under Apps, if it is not connected yet.
• 3. Give the agent the Actions it needs under Access. It starts with none.
• 4. Create an Agent Credential for it (Settings, API keys) and store it as CONTRO1_API_KEY in your secret manager.
• 5. Call Actions from your tools as below. The same credential creates approval requests for work your own code does.

Connect an agent: every path, in full

Run an application Action from a CrewAI tool

Contro1 holds the account and makes the call, so the record is what Contro1 observed. The tool returns what the Action produced; when a person has to approve first, it waits and then returns the result. It never re-submits: a retry could send a second email.

Before writing the input, read the exact input_schema with get_action_contract, or from the Action on the Access page.

import os, uuid
from centcom import CentcomClient, needs_human_resolution

contro1 = CentcomClient(api_key=os.environ["CONTRO1_API_KEY"])  # an Agent Credential

def run_action(action_id: str, input: dict, *, account_mode: str = "shared", **kw):
    """Run a Contro1 Action and return what it produced. Never retries a send."""
    out = contro1.actions.invoke(
        action_id, input,
        authority_mode=kw.pop("authority_mode", "agent_principal"),
        account_mode=account_mode,
        idempotency_key=kw.pop("idempotency_key", str(uuid.uuid4())),
        **kw,
    )
    inv = out["invocation"]
    if inv["state"] == "executed":
        return out["result"]
    if inv["state"] == "awaiting_approval":
        settled = contro1.actions.wait_for_invocation(inv["invocation_id"])
        if needs_human_resolution(settled):
            raise RuntimeError("Outcome unknown; a person must check. Do not retry.")
        return contro1.actions.get_result(settled["invocation_id"])
    raise RuntimeError(f"Not run: {inv['state']} {out.get('not_run', '')}")

from crewai.tools import tool

@tool("List recent emails")
def list_recent_emails(max_results: int = 5) -> list:
    """List the most recent emails in the team mailbox."""
    return run_action("gmail.message.list", {"max_results": max_results})

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 a run trace, the tools you invoked, and the context you retrieved. Each field is optional; add what you have. Identity always comes from your credential. When a named part of a multi-agent system made the call, say which with the Contro1-Sub-Agent header (as_sub_agent in the SDKs): Contro1 registers it under this agent, and it never has more authority than this agent.

• 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. A hook can report every tool call for you.
• 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.
billing = client.as_sub_agent("billing")   # optional: a named part of this agent
billing.create_protocol_request({
    "request_type": "approval",
    "title": "Refund $4,200 to customer 8831",
    "source": {"integration": "crewai"},
    "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