> For the complete documentation index, see [llms.txt](https://docs.kyvvu.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.kyvvu.com/integrations/crewai.md).

# CrewAI

The `kind` field identifies each recorded action kind, such as `step.message` for the final response and `task.error` for a failed task.

## Install

```bash
pip install kyvvu-crewai
```

This installs the Kyvvu SDK and a tested CrewAI release (`crewai>=1.15.21,<2`) into one environment; they do not conflict. It needs **Python 3.10 to 3.13**, because CrewAI does not support Python 3.14 yet.

### `mcp` and CVE-2026-59951

Every CrewAI release up to 1.15.28 pins `mcp~=1.28.1`, so the install above brings in `mcp` 1.28.1, and a dependency audit (`pip-audit`, Dependabot) flags it for CVE-2026-59951 (GHSA-84m7-p3x7-pcfv): an `mcp` *server* below 1.30.0 can be made to grow its memory without bound by a client over Streamable HTTP. CrewAI uses only `mcp`'s client side, and neither Kyvvu nor CrewAI runs an MCP server for you, so a crew is not exposed through it. kyvvu-crewai keeps CrewAI's pin rather than overriding it in its own metadata.

To have the audit clean anyway, replace the pin with a release of the same 1.x line, as Kyvvu's own demo image does. With uv, in your project's `pyproject.toml`:

```toml
[tool.uv]
override-dependencies = ["mcp>=1.30.0,<2"]
```

or, without a project, `uv pip install kyvvu-crewai --override overrides.txt` with `mcp>=1.30.0,<2` in `overrides.txt`. Either resolves `mcp` 1.30.0 next to `crewai` 1.15.28. `uv pip check` then reports the pin you overrode, which is expected: crewai requires `mcp~=1.28.1`, but 1.30.0 is installed.

pip has no override. A constraint (`pip install -c constraints.txt` with `mcp>=1.30.0,<2`) does not work: it narrows CrewAI's pin instead of replacing it, and pip stops with `ResolutionImpossible`. Upgrade after the install instead:

```bash
pip install kyvvu-crewai
pip install "mcp>=1.30.0,<2"
```

The second command prints `crewai 1.15.28 requires mcp~=1.28.1, but you have mcp 1.30.0 which is incompatible.` and installs it; `pip check` keeps reporting the same pair. Drop the override once a CrewAI release allows `mcp>=1.30`.

## With the platform

Build the crew as usual. This one has one agent with one tool, `search_web`, and a real model (CrewAI reads the provider key, for example `OPENAI_API_KEY`, from the environment):

```python
from crewai import Agent, Crew, Task
from crewai.tools import tool


@tool("search_web")
def search_web(query: str) -> str:
    """Search the web."""
    ...


researcher = Agent(
    role="Researcher",
    goal="Draft briefings on AI governance from public sources",
    backstory="You research public sources.",
    llm="gpt-4o-mini",
    tools=[search_web],
)
task = Task(
    description="Write a short briefing on {topic}.",
    expected_output="A one-paragraph briefing",
    agent=researcher,
)
crew = Crew(agents=[researcher], tasks=[task])
```

Then register the agent and run the crew through the listener:

```python
from kyvvu import Kyvvu
from kyvvu.exceptions import KyvvuBlockedError
from kyvvu.schemas import Environment
from kyvvu_crewai import KyvvuCrewAIListener

kv = Kyvvu(
    api_url="https://platform.kyvvu.com",
    api_key="KvKey-...",
    agent_key="research-crew",
    # Left out (and KV_ENVIRONMENT unset), the agent registers as production.
    environment=Environment.DEVELOPMENT,
)
kv.register_agent(
    name="Research Crew",
    purpose="Researches public sources and drafts briefings on AI governance",
    # The person or team accountable for the agent.
    owner_id="ai-governance@example.com",
    # Every tool the crew can call, by its CrewAI tool name.
    declared_tools=["search_web"],
)
listener = KyvvuCrewAIListener(kv)

try:
    result = listener.kickoff(crew, inputs={"topic": "AI governance"})
except KyvvuBlockedError as exc:
    print(f"Policy blocked: {exc}")
```

**`declared_tools` matters.** Every registered agent gets the [bundled baseline](/policy-authoring/bundled-baseline.md), whose allowlist policy blocks every tool call whose name is not on that list, and blocks all of them when the list is empty. List the name of each tool your crew uses. A tool named with spaces or capitals is recorded under CrewAI's sanitized name (`"Search Web"` becomes `search_web`), and that is the name to declare. The baseline also wants a `purpose` of at least 30 characters, and warns at registration without one.

**`owner_id` and `environment`.** The baseline also asks every agent to name an accountable owner: without `owner_id`, registration reports "Agent must have a designated owner" (a `medium` warning, risk 0.50, not a block) and the platform opens an incident for it, as [Your first agent](/getting-started/first-agent.md#why-the-demo-agent-shows-compliance-warnings) explains. The value is checked whether or not the platform stores it (`store_metadata`). Without `environment=` an agent registers as `KV_ENVIRONMENT` says, and as `production` when that is unset; the snippet registers a development agent, and a production deployment passes `Environment.PRODUCTION`.

## Offline, with no account

No API key and no network are needed to enforce the bundled baseline. Give the client an `agent_id`, declare the crew's tools, and pass the baseline as `policies=`. This script runs as written: a scripted model stands in for a real one, so it needs no provider key either.

```python
"""Offline: a CrewAI crew governed by the bundled baseline, no account."""
from crewai import Agent, Crew, Task
from crewai.llms.base_llm import BaseLLM
from crewai.tools import tool

from kyvvu import Kyvvu, KyvvuBlockedError
from kyvvu.baseline import load_baseline_policies
from kyvvu.schemas import RiskClassification
from kyvvu_crewai import KyvvuCrewAIListener


class ScriptedLLM(BaseLLM):
    """Replays scripted replies; stands in for a real model."""

    replies: list[str]
    calls: int = 0

    def call(self, messages, **kwargs):
        self.calls += 1
        return self.replies[min(self.calls, len(self.replies)) - 1]


@tool("search_web")
def search_web(query: str) -> str:
    """Search the web."""
    return "Article 50 of the EU AI Act sets transparency obligations."


llm = ScriptedLLM(model="scripted", replies=[
    'Thought: I should search.\nAction: search_web\nAction Input: {"query": "EU AI Act Article 50"}',
    "Thought: I know the answer.\nFinal Answer: Article 50 sets transparency obligations.",
])
researcher = Agent(
    role="Researcher", goal="Answer questions about AI governance",
    backstory="You research public sources.", llm=llm, tools=[search_web],
)
task = Task(
    description="What does {topic} require?", expected_output="One sentence",
    agent=researcher,
)
crew = Crew(agents=[researcher], tasks=[task])

kv = Kyvvu(
    agent_id="research-crew",
    risk_classification=RiskClassification.LIMITED,
    declared_tools=["search_web"],
    read_only_tools=["search_web"],
    policies=load_baseline_policies(),
    log_location="none",  # "stdout" (the offline default) prints the trace
)
listener = KyvvuCrewAIListener(kv)

try:
    result = listener.kickoff(crew, inputs={"topic": "EU AI Act Article 50"})
    print(result)
except KyvvuBlockedError as exc:
    print(f"Policy blocked: {exc}")
```

Saved as `crew.py` and run with `python crew.py`, it prints (timestamps trimmed):

```
kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO policy violation: task=42a8e413-0e3 step='crew_output' action=warn risk=0.75 (2 violations)
/path/to/crew.py:54: UserWarning: Policy warning on 'crew_output': risk_score=0.75
  result = listener.kickoff(crew, inputs={"topic": "EU AI Act Article 50"})
kyvvu_engine.io.runner INFO task 42a8e413-0e3 ended — 7 steps (task.start×1, step.message×2, step.model×2, step.resource×1, task.end×1)
Article 50 sets transparency obligations.
```

The tool ran, because it is declared, and the crew's answer came back. The warning is explained [below](#why-every-run-warns-on-crew_output), and [a gate recorded from a task guardrail](#recording-a-gate-from-inside-the-crew) clears it.

* **`agent_id` is required** offline; `agent_key` is not. Without platform credentials nothing registers, so the baseline's registration-time policies are inert and its step policies still enforce.
* `declared_tools=` can also come from `KV_DECLARED_TOOLS=search_web`, and the trace location from `KV_LOG_LOCATION` (a file path keeps it; `KV_LOG_FORMAT=json` for JSON lines). Set `log_location="stdout"` or leave it out to see the trace.
* For a manifest of your own, set `KV_POLICY_FILE=my-manifest.yaml` instead of passing `policies=`.

### When a policy blocks a call

Run the same script with `search_web` taken out of the list (`declared_tools=[]`), and the baseline's allowlist blocks the tool before its body runs. `listener.kickoff()` raises `KyvvuBlockedError`, which the script catches:

```
kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO policy violation: task=7c06bb75-d81 step='search_web' action=block risk=1.00 (1 violations)
kyvvu_engine.io.runner INFO task 7c06bb75-d81 ended — 5 steps (task.start×1, step.message×1, step.model×1, step.resource×1, task.error×1)
Policy blocked: Step 'search_web' blocked (risk_score=1.00): Tool calls must be in the declared allowlist (manifest OWASP Top 10 for Agentic Applications - Default Security Template)
```

The block is reported once, by the engine's `policy violation` line, and the exception carries it to your code. CrewAI itself would also log the abort as `crewai.flow.runtime ERROR Error executing listener call_llm_and_parse: Step 'search_web' blocked ...` and print two `[CrewAIEventsBus] Warning: Event pairing mismatch` lines while it unwinds the crew; the listener quiets those three for a Kyvvu block only. Any other failure is left as CrewAI reports it: a model that raises, for example, logs that ERROR line once per attempt (CrewAI retries a failed task up to the agent's `max_retry_limit`, so with the default of 2 it appears three times), prints no pairing warnings, and `kickoff()` raises the model's own exception. On the [direct `crew.kickoff()` path](#calling-crewkickoff-directly) the ERROR line is quiet too, but the two pairing warnings still print, and CrewAI raises its own `HookAborted` with the same message rather than `KyvvuBlockedError`.

## Why every run warns on `crew_output`

`listener.kickoff()` records the crew's final answer as a `step.message POST` named `crew_output`. Under the bundled baseline that step is a **compliance finding** unless the run records a gate before it: two EU AI Act Article 50 policies warn (`high`, risk 0.75), because the model's answer reaches a person with no gate recorded after the model call. The content is not marked as AI-generated (Art. 50(2)), and the person was not told they are dealing with an AI (Art. 50(1)). It is a warning, not a block: the crew's result is returned, the verdict is recorded on the step (`meta["kyvvu.eval"].action` is `warn`), and with the platform each violated policy opens an incident.

The remedy is a passing `step.gate` recorded after the last model call and before the answer is delivered: the point where your application marks the content and discloses that it comes from an AI. In a crew, record it with `listener.gate()` from a [task guardrail](#recording-a-gate-from-inside-the-crew), which CrewAI runs at exactly that point. The same warning on a decorator agent is explained in [Your first agent](/getting-started/first-agent.md#why-the-demo-agent-shows-compliance-warnings).

The `UserWarning` names the line in **your** code that called `listener.kickoff()`, not a line inside Kyvvu or CrewAI. Python shows it once per location; `warnings.simplefilter("error", UserWarning)` turns it into an exception.

## Recording a gate from inside the crew

`listener.gate(name, *, decision, kind="human", **properties)` records a completed `step.gate` in the task of the kickoff that is running:

* `decision` is `"pass"` or `"deny"`, recorded as `output.decision`.
* `kind` is `"human"` (a person decided) or `"guardrail"` (code decided), recorded as `properties.kind`.
* Any further keyword is recorded as a property of the gate (I-JSON values only, because policies read them). The CrewAI task description is added under `properties.crewai`.

Call it from code CrewAI runs during the kickoff: a task guardrail, a task callback, or a tool. CrewAI runs that code with no Kyvvu task selected, so `kv.add_step()` cannot record there; `listener.gate()` finds the kickoff the same way the listener attributes the crew's tool and model calls, so two kickoffs running at once each record into their own task. Called anywhere else, it raises `KyvvuConfigError`. A `kind` or `decision` outside the values above raises `ValueError`, as `kv.add_step()` does.

A task guardrail runs after the agent's last model call and before the crew returns its answer, which is where the baseline wants the gate. This script marks the answer and records the gate there, and runs as written with no account, network or provider key:

```python
"""Offline, with a disclosure gate: no Art. 50 warning on crew_output."""
from crewai import Agent, Crew, Task
from crewai.llms.base_llm import BaseLLM

from kyvvu import Kyvvu
from kyvvu.baseline import load_baseline_policies
from kyvvu.schemas import RiskClassification
from kyvvu_crewai import KyvvuCrewAIListener

DISCLOSURE = "\n\n(This answer was generated by an AI system.)"


class ScriptedLLM(BaseLLM):
    """Replays scripted replies; stands in for a real model."""

    replies: list[str]
    calls: int = 0

    def call(self, messages, **kwargs):
        self.calls += 1
        return self.replies[min(self.calls, len(self.replies)) - 1]


kv = Kyvvu(
    agent_id="briefing-crew",
    risk_classification=RiskClassification.LIMITED,
    policies=load_baseline_policies(),
    log_location="none",
)
listener = KyvvuCrewAIListener(kv)


def disclose(output):
    """Task guardrail: mark the answer as AI-generated, then record the gate."""
    output.raw += DISCLOSURE
    listener.gate("ai_disclosure", kind="guardrail", decision="pass", marking="appended notice")
    return True, output


llm = ScriptedLLM(model="scripted", replies=[
    "Thought: I know the answer.\nFinal Answer: Article 50 sets transparency obligations.",
])
writer = Agent(
    role="Writer", goal="Answer questions about AI governance",
    backstory="You write short briefings.", llm=llm,
)
task = Task(
    description="What does {topic} require?", expected_output="One sentence",
    agent=writer, guardrail=disclose,
)
crew = Crew(agents=[writer], tasks=[task])

result = listener.kickoff(crew, inputs={"topic": "EU AI Act Article 50"})
print(result)
```

Saved as `crew_gate.py` and run with `python crew_gate.py`, it prints (timestamps trimmed):

```
kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO task 15c01ba9-12b ended — 6 steps (task.start×1, step.message×2, step.model×1, step.gate×1, task.end×1)
Article 50 sets transparency obligations.

(This answer was generated by an AI system.)
```

No `policy violation` line and no `UserWarning`: the trace reads `step.model`, then the `ai_disclosure` gate, then `crew_output`, and both Article 50 policies pass on `crew_output`.

**A `deny` decision is a record, not a block.** `listener.gate(..., decision="deny")` does not raise and does not stop the crew. A denied gate does not satisfy the disclosure policy, so `crew_output` still warns (the marking policy only asks that some gate sits between the model call and the delivery). A guardrail that wants CrewAI to retry the task, or to stop, must also return CrewAI's failure result:

```python
def review(output):
    approved = reviewer_approves(output.raw)  # your own check
    listener.gate("human_review", kind="human", decision="pass" if approved else "deny")
    return (True, output) if approved else (False, "Rejected by the reviewer")
```

## How enforcement works

`listener.kickoff()` runs the crew inside a Kyvvu run scope. It evaluates the complete final response as `step.message POST` in the calling thread before returning the original CrewAI output. A blocked response is recorded once as a blocked step, closes the task as `task.error`, and raises `KyvvuBlockedError`. Tool and model calls are enforced earlier by CrewAI's synchronous before-call hooks, and a blocked tool or model call also reaches the caller as `KyvvuBlockedError`.

### Calling `crew.kickoff()` directly

Tool and model calls are enforced the same way, but **the final answer is recorded without being evaluated**. CrewAI's event bus cannot carry a block back to the caller of `crew.kickoff()`, so the `crew_output` step carries no verdict on this path: no `policy violation` line, no `UserWarning`, and no incident, whatever an outbound message policy says. When a loaded step policy could apply to that message, the listener says so once per run:

```
kyvvu_crewai.crewai WARNING CrewAI direct crew.kickoff() records the final answer without evaluating it against the loaded step policies; use listener.kickoff(crew, inputs=...) to enforce them
```

Use `listener.kickoff()` whenever the final answer must be governed.