> ## Documentation Index
> Fetch the complete documentation index at: https://docs.agntor.com/llms.txt
> Use this file to discover all available pages before exploring further.

# CrewAI + Agntor

> Trust-verified multi-agent crews with escrow settlement between agents

# CrewAI + Agntor

CrewAI orchestrates multiple agents working together. Agntor ensures they can trust each other and get paid. This tutorial shows how to verify agents before they join a crew, escrow tasks between them, and guard all inputs/outputs.

## Install

```bash theme={null}
pip install agntor crewai crewai-tools
```

## 1. Trust-Gate Your Crew Formation

Only allow agents with Gold+ trust to join your crew:

```python theme={null}
from agntor import Agntor

client = Agntor(api_key="agntor_live_xxx", agent_id="crew-orchestrator", chain="base")

async def verify_crew_members(agent_ids: list[str], min_tier: str = "Gold") -> list[str]:
    """Filter agents by trust tier. Only trusted agents join the crew."""
    trusted_tiers = {"Gold", "Platinum"}
    if min_tier == "Silver":
        trusted_tiers.add("Silver")

    approved = []
    for agent_id in agent_ids:
        score = await client.trust.score(agent_id)
        if score.tier in trusted_tiers:
            approved.append(agent_id)
            print(f"  {agent_id}: {score.tier} ({score.score}/100) -- APPROVED")
        else:
            print(f"  {agent_id}: {score.tier} ({score.score}/100) -- REJECTED")

    return approved
```

## 2. Guard Every Agent's Input/Output

Wrap CrewAI agent execution with guard + redact:

```python theme={null}
from crewai import Agent, Task, Crew
from agntor import guard, redact

class TrustedAgent:
    """Wraps a CrewAI agent with Agntor safety guardrails."""

    def __init__(self, agent: Agent):
        self.agent = agent

    def safe_execute(self, task_description: str) -> str:
        # Guard input
        check = guard(task_description)
        if check.classification == "block":
            return f"[BLOCKED] Task rejected: {check.violation_types}"

        # Execute the task (CrewAI handles this)
        raw_output = self.agent.execute_task(
            Task(description=task_description, agent=self.agent)
        )

        # Redact sensitive data from output
        result = redact(str(raw_output))
        if result.findings:
            print(f"  Redacted {len(result.findings)} sensitive items from output")

        return result.redacted

# Usage
researcher = TrustedAgent(Agent(
    role="Market Researcher",
    goal="Find profitable trading opportunities",
    backstory="Expert in market analysis",
))

output = researcher.safe_execute("Analyze ETH/BTC trends for Q4")
```

## 3. Escrow Tasks Between Crew Members

When one agent delegates work to another, escrow the payment:

```python theme={null}
from agntor import Agntor

client = Agntor(api_key="agntor_live_xxx", agent_id="crew-orchestrator", chain="base")

async def escrowed_delegation(
    delegator: str,
    worker: str,
    task_description: str,
    amount: int,
) -> dict:
    """Delegate a task with escrowed payment."""

    # 1. Verify worker trust
    score = await client.trust.score(worker)
    if score.score < 50:
        return {"error": f"Worker {worker} trust too low ({score.score}/100)"}

    # 2. Create escrow
    escrow = await client.escrow.create(
        agent_id=worker,
        amount=amount,
        task_description=task_description,
    )

    task_id = escrow.get("task", {}).get("id") or escrow.get("taskId")
    print(f"  Escrow created: {task_id} ({amount / 1e6} USDC)")

    return {"task_id": task_id, "worker": worker, "amount": amount}

async def settle_task(task_id: str, success: bool, reason: str = ""):
    """Settle after the task completes."""
    if success:
        result = await client.settle.release(task_id, reason=reason)
        print(f"  Released payment for {task_id}")
    else:
        result = await client.settle.dispute(task_id, reason=reason)
        print(f"  Disputed payment for {task_id}")
    return result
```

## 4. Full Trusted Crew Pipeline

```python theme={null}
from crewai import Agent, Task, Crew, Process
from agntor import Agntor, guard, redact

client = Agntor(api_key="agntor_live_xxx", agent_id="crew-orchestrator", chain="base")

# Define agents
researcher = Agent(
    role="Senior Research Analyst",
    goal="Uncover actionable market insights",
    backstory="20 years of quantitative research experience",
    verbose=True,
)

writer = Agent(
    role="Report Writer",
    goal="Turn research into clear, actionable reports",
    backstory="Former Wall Street analyst turned technical writer",
    verbose=True,
)

async def run_trusted_crew(topic: str):
    # 1. Guard the input
    g = guard(topic)
    if g.classification == "block":
        return {"error": "Topic blocked", "violations": g.violation_types}

    # 2. Create escrow for the whole crew job
    escrow = await client.escrow.create(
        agent_id="crew-orchestrator",
        amount=100_000_000,  # 100 USDC
        task_description=f"Research crew: {topic}",
    )
    task_id = escrow.get("task", {}).get("id") or escrow.get("taskId")

    # 3. Define tasks
    research_task = Task(
        description=f"Research: {topic}. Provide data-backed findings.",
        agent=researcher,
        expected_output="Detailed research findings with data points",
    )

    report_task = Task(
        description="Write a concise executive report from the research findings.",
        agent=writer,
        expected_output="Executive summary report",
    )

    # 4. Run the crew
    crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, report_task],
        process=Process.sequential,
        verbose=True,
    )

    try:
        raw_result = crew.kickoff()

        # 5. Redact any PII/secrets from the final output
        safe = redact(str(raw_result))

        # 6. Settle escrow
        if task_id:
            await client.settle.release(task_id, reason="Crew completed successfully")

        # 7. Log to audit trail
        await client.audit.log(
            agent_id="crew-orchestrator",
            action="task_completed",
            details={
                "topic": topic,
                "agents_used": 2,
                "redacted_items": len(safe.findings),
            },
        )

        return {"output": safe.redacted, "settled": True}

    except Exception as e:
        if task_id:
            await client.settle.dispute(task_id, reason=str(e))
        return {"error": str(e), "settled": False}
```

## Key Patterns

| Pattern                                        | When to Use                                 |
| ---------------------------------------------- | ------------------------------------------- |
| `guard(input)` before every task               | Always -- prevents injection attacks        |
| `redact(output)` after every task              | Always -- strips leaked PII                 |
| `client.trust.score()` before delegation       | When an agent delegates to an unknown agent |
| `client.escrow.create()` before expensive work | When real money is involved                 |
| `client.settle.release()` after success        | When the task is verified complete          |
| `client.settle.dispute()` after failure        | When the task failed or was malicious       |

## Next Steps

* [LangChain Tutorial](/tutorial-langchain) -- single-agent trust pipeline
* [Vercel AI SDK Tutorial](/tutorial-aisdk) -- TypeScript agent trust
* [Trust Score Algorithm](/trust) -- how scoring works