Skip to content

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

pip install kyvvu-crewai

This installs the kyvvu Python client and a CrewAI release from kyvvu-crewai's range (crewai>=1.15.21,<2; Kyvvu's tests run 1.15.32) into one environment; they do not conflict. It needs Python 3.11 to 3.13, because CrewAI does not support Python 3.14 yet.

mcp and CVE-2026-59951

CrewAI 1.15.29 and later require mcp>=1.30.0,<2, and the tested lock uses CrewAI 1.15.32 with mcp 1.30.0, which carries the 1.x fix for CVE-2026-59951 (GHSA-84m7-p3x7-pcfv). The install above therefore needs no override: pip and uv resolve the fixed mcp from CrewAI's own metadata, and the environment checks clean:

pip check

which prints No broken requirements found.

CrewAI 1.15.21 to 1.15.28, the older end of kyvvu-crewai's range, pin mcp~=1.28.1, which is affected: in those versions a stateful Streamable HTTP server can retain unlimited session state. An environment that already holds one of those releases keeps that mcp until CrewAI itself is upgraded to 1.15.29 or later. If you added an override-dependencies entry or a post-install mcp upgrade to work around the old pin, you can remove it.

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). The search tool returns example content; replace its body with your search provider:

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


@tool("search_web")
def search_web(query: str) -> str:
    """Return example content; replace with your search provider."""
    return f"example search result for {query}: a public AI governance briefing"


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:

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, 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 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.

"""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]

    def get_input_tokens_count(self, messages, tools=None, response_model=None):
        return 0  # Scripted replies use no tokens; CrewAI warns if it must estimate.


@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 and warning locations omitted; task IDs shown as <task-id>):

kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO policy violation: task=<task-id> step='crew_output' action=warn risk=0.75 (2 violations)
UserWarning: Policy warning on 'crew_output': risk_score=0.75
kyvvu_engine.io.runner INFO task <task-id> 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, and a gate recorded from a task guardrail 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=<task-id> step='search_web' action=block risk=1.00 (1 violations)
kyvvu_engine.io.runner INFO task <task-id> 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 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, which CrewAI runs at exactly that point. The same warning on a decorator agent is explained in Your first agent.

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:

"""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]

    def get_input_tokens_count(self, messages, tools=None, response_model=None):
        return 0  # Scripted replies use no tokens; CrewAI warns if it must estimate.


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 omitted; task IDs shown as <task-id>):

kyvvu_engine.io.runner INFO Loaded 15 policies from the local policy source (inline policies=).
kyvvu_engine.io.runner INFO task <task-id> 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:

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, and the final answer is evaluated too, but it cannot be withheld. CrewAI's event bus delivers the answer to the caller of crew.kickoff() whatever the handler decides, so a blocking outbound message policy records the crew_output step as blocked, files an incident, ends the task with task.error and sets listener.is_blocked, while the caller still receives the answer. When a loaded step policy could apply to that message, the listener says so once per run:

kyvvu_crewai.crewai WARNING CrewAI's event bus cannot stop direct crew.kickoff() from delivering a blocked final answer; use listener.kickoff(crew, inputs=...) to withhold it

Use listener.kickoff() whenever a blocked final answer must not reach the caller.