A safe CrewAI human-in-the-loop boundary lets the crew prepare a proposal but keeps the real side effect in ordinary application code. Create one idempotent Actionbox request for the run, wait for a typed decision, fail closed on rejection or timeout, and report the operation's outcome after an approved run finishes.
CrewAI also has native human feedback. Its @human_feedback flow decorator can pause a flow, persist pending state, and resume through a feedback provider. Use that when feedback is part of the flow itself. The pattern below is narrower: a hard approval boundary before one consequential operation.
Why separate the proposal from the operation?
An agent is useful for assembling evidence, comparing options, and writing a proposed plan. It should not gain deployment, payment, or deletion authority merely because its plan reads well.
Keep ownership explicit:
| Concern | Owner |
|---|---|
| Generate the proposal | CrewAI |
| Store and present the decision | Actionbox |
| Hold the operation's credentials | Your application |
| Execute or refuse the operation | Your application |
| Record whether execution worked | Actionbox outcome |
Install CrewAI and the Actionbox SDK
Both packages are published on PyPI:
python -m pip install actionbox-sdk crewai
export ACTIONBOX_API_KEY="axb_live_..."Create a Source in Actionbox and save the key when it is shown. Keep the key in your runtime secret store, not in the script or the proposal.
Build a fail-closed approval adapter
The adapter rejects oversized proposals instead of silently truncating them. A truncated plan can hide the exact command, scope, or rollback detail that would have changed the reviewer's answer.
from __future__ import annotations
import hashlib
import time
from typing import Any, Callable
MAX_PROPOSAL_BYTES = 16_000
def request_approval(client, *, run_id: str, proposal: Any, timeout=3600):
rendered = str(proposal)
if len(rendered.encode("utf-8")) > MAX_PROPOSAL_BYTES:
raise ValueError("proposal is too large for a complete approval request")
if not run_id or len(run_id.encode("utf-8")) > 255:
raise ValueError("run_id must be between 1 and 255 UTF-8 bytes")
identity = hashlib.sha256(run_id.encode()).hexdigest()
action = client.create(
title="Approve the CrewAI proposal?",
description="CrewAI prepared this plan. Review it before the operation runs.",
priority="high",
interaction={
"type": "boolean",
"label": "Run this proposal?",
"true_label": "Approve",
"false_label": "Reject",
},
context=[
{
"type": "key_value",
"title": "Crew run",
"items": {"Run ID": run_id, "Framework": "CrewAI"},
},
{
"type": "code",
"title": "Proposed work - remove secrets before sending",
"language": "text",
"content": rendered,
},
],
metadata={
"origin": {"provider": "agent", "ref": f"CrewAI - {run_id}"[:160]},
"framework": "crewai",
"run_id": run_id,
},
idempotency_key=f"crewai:{identity}",
)
response = action.wait(timeout=timeout)
if response is None and action.status == "open":
client.cancel(
action.id,
"CrewAI approval timed out; operation rejected fail-closed.",
)
approved = (
isinstance(response, dict)
and response.get("type") == "boolean"
and response.get("value") is True
)
return action, approved
def execute_after_approval(
action,
approved: bool,
operation: Callable[[], Any],
):
if not approved:
return None
started = time.monotonic()
try:
result = operation()
except Exception:
action.report_outcome(
"failed",
duration_ms=int((time.monotonic() - started) * 1000),
reason_code="operation_failed",
)
raise
action.report_outcome(
"success",
duration_ms=int((time.monotonic() - started) * 1000),
)
return resultThe stable run_id is part of the safety model. If the process loses the create response and retries, Actionbox returns the same logical request. Use an ID from your own job, deployment, or transaction record. Do not create a random ID on each retry.
Put it around a CrewAI proposal
This crew can plan a release, but it has no deployment tool and no production credential:
import os
from actionbox import Actionbox
from crewai import Agent, Crew, Process, Task
planner = Agent(
role="Release planner",
goal="Prepare a bounded release plan for the supplied version",
backstory="You plan releases, but you never perform the deployment.",
)
task = Task(
description="Prepare a release plan for version {version} in staging.",
expected_output="A short plan with target, checks, risks, and rollback steps.",
agent=planner,
)
crew = Crew(agents=[planner], tasks=[task], process=Process.sequential)
proposal = crew.kickoff(inputs={"version": "2.19.0"})
with Actionbox(os.environ["ACTIONBOX_API_KEY"]) as actionbox:
action, approved = request_approval(
actionbox,
run_id="release-2.19.0-staging",
proposal=proposal,
)
result = execute_after_approval(
action,
approved,
# Replace this with an idempotent deployment operation.
lambda: "Deployment would run here",
)
print(result or "Proposal rejected or timed out; nothing ran.")The real operation belongs inside execute_after_approval, not in a CrewAI task that ran before the decision. Give that operation its own idempotency key too. An approved process can still restart between executing the side effect and saving its result.
Handle every terminal path
| Path | What the application does |
|---|---|
| Approved | Execute once, then report success or failure |
| Rejected | Return without executing |
| Wait timeout | Cancel the still-open Action and return without executing |
| Actionbox unavailable | Raise and stop before the operation |
| Proposal too large | Reject it before creating an incomplete review |
| Operation raises | Report a failed outcome and preserve the exception |
| Process restarts | Reuse the same run ID and Action idempotency key |
Do not turn an API error into an approval. Do not ask the model to infer whether silence means yes. A missing decision is not a decision.
What should the reviewer see?
The proposal should answer the questions needed for this operation, not dump an agent transcript into the inbox. For a release, that usually means:
- the exact target and version;
- evidence from tests or staging;
- affected services or customers;
- known risks and the rollback step.
Remove secrets before the Action is created. A warning in the UI cannot make an exposed credential safe.
When to use CrewAI's feedback provider instead
CrewAI's human feedback system is a better fit when feedback should revise an earlier flow method, emit one of several flow outcomes, or resume a persisted flow from another process. Its custom HumanFeedbackProvider interface can connect an external delivery mechanism to that lifecycle.
Use the adapter in this guide when the contract is simpler: CrewAI proposes, a person authorizes or refuses, and application code owns the protected operation. Keeping that boundary small makes retries and failure behavior easier to test.
Test before connecting a real tool
Run the example first with a harmless function. Verify approval, rejection, timeout, duplicate create, oversized context, execution failure, and outcome reporting. Only then replace the placeholder with the real idempotent operation.
For the same boundary in other agent runtimes, see the human-in-the-loop AI agent guide. For restart and state ownership patterns, read durable execution explained.
