AgentTrust Docs

Overview

The CrewAI adapter provides AgentTrustCallback — a callback handler attached to CrewAI tasks that validates agent output after each task execution.

Tier: Team+

Why It Matters

CrewAI multi-agent crews need per-task governance so individual agent outputs are validated before being passed to downstream crew members.

Prerequisites

pip install "agentrust-py[crewai]"
# Team+ API key required

Step-by-Step Guide

The tier gate raises `RuntimeError`, and only sometimes

This adapter calls _check_adapter_tier() at construction, which raises a plain RuntimeError prefixed [AgentTrust] — except TierGateError will not catch it.

The check also short-circuits: it returns without checking anything when AGENTRUST_ENV is development, dev, demo, or test, when no api_key argument was passed to the constructor, or when the key does not start with at_. Constructing the adapter with no key therefore never gates — it will run and simply be rejected by the gateway later.

1. Import callback

from agentrust_sdk.adapters.crewai import AgentTrustCallback

2. Attach to crew task

CrewAI's Task takes a single callback callable — pass the adapter's on_task_complete method, not the object and not a list.

from crewai import Agent, Task, Crew

callback = AgentTrustCallback(
    agent_id="crew-researcher",
    base_url="http://127.0.0.1:8765",   # adapter default is :8000
)

task = Task(
    description="Research AI governance trends",
    agent=researcher,
    callback=callback.on_task_complete,   # singular, and a callable
)
crew = Crew(agents=[researcher], tasks=[task])
result = crew.kickoff()

on_task_complete returns the task output unchanged, so it is safe to drop into an existing pipeline.

3. Configure block behaviour

block_on_block defaults to True: a block decision raises BlockedError out of the callback and aborts the crew run. block_on_review defaults to False.

callback = AgentTrustCallback(
    agent_id="crew-researcher",
    block_on_block=True,    # abort the crew on a block
    block_on_review=False,  # let escalate / request_evidence through
    user="crewai",          # identity recorded on the envelope
)

Chain your own callback with task_callback(), and read the last verdict with callback.last_result:

task = Task(..., callback=callback.task_callback(my_existing_callback))
print(callback.last_result.decision.outcome)

Examples

See examples/crewai_agent.py.

Best Practices

Attach callback to every task that produces user-facing output
Use distinct agent_id per crew role (researcher, writer, reviewer)
Run embedded gateway locally; remote gateway in production
Combine with trust chain for multi-crew pipelines (Enterprise)

Common Mistakes

Single agent_id for all crew members (loses per-role audit granularity)
Catching TierGateError around construction — the adapter raises plain RuntimeError
Not handling task failures when governance blocks output

Troubleshooting

Issue	Fix
`RuntimeError: [AgentTrust] CrewAI adapter requires Team tier…`	Upgrade to Team tier (this is not a `TierGateError`)
Callback not firing	Use `callback=cb.on_task_complete` (singular, a callable)
Crew continues after block	Confirm `block_on_block=True` and that the gateway is reachable

CrewAI Integration