Skip to content

kyvvu_crewai Python API

CrewAI listener for Kyvvu.

kyvvu_crewai.KyvvuCrewAIListener

KyvvuCrewAIListener(kyvvu: 'Kyvvu', template: object = default_template, extract_tools: bool = False, crew: Any = None) -> None

CrewAI listener with synchronous call and final-message enforcement.

Subscribes to CrewAI's event bus and maps events to the v0.05 behavioral vocabulary. One Kyvvu task per crew kickoff. Agent and Task events are metadata enrichments on subsequent steps.

Parameters:

  • kyvvu ('Kyvvu') – A configured Kyvvu instance.
  • template (object) – A BehaviorTemplate (replaces the built-in and governs both steps and task boundaries), or the default default_template. None is refused.
  • extract_tools (bool) – When True and a crew is provided, inspect the crew's agents for their tools and pass them to register_agent(declared_tools=[...]).
  • crew (Any) – Optional CrewAI Crew instance used for tool extraction.

Example::

kv = Kyvvu(api_url="...", api_key="KvKey-...", agent_key="bot")
kv.register_agent(name="Research Crew")
listener = KyvvuCrewAIListener(kv, extract_tools=True, crew=my_crew)
result = listener.kickoff(my_crew)

Functions:

  • build_properties – Inject properties.crewai metadata BEFORE policy evaluation.
  • gate – Record a completed step.gate in the running kickoff's task.
  • kickoff – Run a crew and enforce its final message before returning it.
  • on_block – Retain the original policy block for callers and later hooks.
  • run_scope – Keep a kickoff's policy decision with its caller, including async callers.
  • setup_listeners – Register all event handlers on the CrewAI event bus.
  • tool_is_declared – Check whether a declared tool has the normalized name.

Attributes:

Attach governance hooks and optionally extract tools from the crew.

kyvvu_crewai.KyvvuCrewAIListener.blocked_exception

blocked_exception: Optional[KyvvuBlockedError]

The KyvvuBlockedError that blocked this task, or None.

A caller can re-raise this after kickoff() to surface a block swallowed by a task or message event handler::

result = crew.kickoff()
if listener.blocked_exception is not None:
    raise listener.blocked_exception

kyvvu_crewai.KyvvuCrewAIListener.blocked_reason

blocked_reason: Optional[str]

The error message from the policy that blocked execution, or None.

kyvvu_crewai.KyvvuCrewAIListener.boundary_contexts

boundary_contexts: Mapping[ActionKind, Sequence[Dict[str, Any]]] = {ActionKind.task_start: ({'event': 'crew_kickoff_started'},), ActionKind.task_end: ({'event': 'crew_kickoff_completed'},), ActionKind.task_error: ({'event': 'crew_kickoff_failed'},)}

kyvvu_crewai.KyvvuCrewAIListener.build_properties

build_properties(matched: Dict[str, Any], context: Dict[str, Any], step_run_id: str) -> Dict[str, Any]

Inject properties.crewai metadata BEFORE policy evaluation.

Overrides the base hook so CrewAI agent/task metadata (carried on the event context) is visible to policy rules at evaluation time.

kyvvu_crewai.KyvvuCrewAIListener.gate

gate(name: str, *, decision: str, kind: str = 'human', **properties: Any) -> None

Record a completed step.gate in the running kickoff's task.

Call it from code CrewAI runs during a kickoff: a task guardrail, a task callback, or a tool. The gate is written into the Kyvvu task of the kickoff that code belongs to, resolved the way the listener attributes that kickoff's tool and model calls, so concurrent kickoffs each record into their own task.

A passing gate after the last model call and before the crew's answer is delivered is what the bundled baseline's two EU AI Act Article 50 policies ask for: the point where your application marks the content as AI-generated and discloses that an AI produced it. A task guardrail runs exactly there::

def disclose(output):
    listener.gate("ai_disclosure", kind="guardrail", decision="pass")
    return True, output

Task(..., guardrail=disclose)

Recording decision="deny" does not stop anything by itself: it is a decision on the record, and a denied gate satisfies no policy that requires a gate. A guardrail that wants CrewAI to retry or stop must also return its failure result, (False, "reason").

Parameters:

  • name (str) – The gate's step name (1-255 characters, no control characters).
  • decision (str) – "pass" or "deny", recorded as output.decision.
  • kind (str) – "human" (a person decided) or "guardrail" (code decided), recorded as properties.kind.
  • **properties (Any) – Further properties recorded on the gate, I-JSON values only (policies read them). The CrewAI agent role and task description are added under crewai when known.

Raises:

  • KyvvuConfigError – If no governed kickoff is running in the caller's context, or its task is no longer open.
  • ValueError – If kind or decision is not one of the values above (the Python client's gate validation, kv.add_step).
  • KyvvuBlockedError – If a policy blocks the gate step itself. The block is kept for the kickoff as any other block is.

kyvvu_crewai.KyvvuCrewAIListener.is_blocked

is_blocked: bool

Whether a policy has blocked this listener.

Tool and model blocks abort through synchronous hooks. A block from a task or message event can still be swallowed by CrewAI's event bus; check this property after crew.kickoff() to detect those blocks::

listener = KyvvuCrewAIListener(kv)
result = crew.kickoff()
if listener.is_blocked:
    print(f"Blocked: {listener.blocked_reason}")
    # Handle the block — e.g. discard the result, alert, etc.

Returns:

  • bool – True if a step was blocked by policy during this task.

kyvvu_crewai.KyvvuCrewAIListener.kickoff

kickoff(crew: Any, inputs: Optional[Dict[str, Any]] = None) -> Any

Run a crew and enforce its final message before returning it.

kyvvu_crewai.KyvvuCrewAIListener.on_block

on_block(task_id: str, exc: KyvvuBlockedError) -> None

Retain the original policy block for callers and later hooks.

kyvvu_crewai.KyvvuCrewAIListener.run_scope

run_scope() -> Iterator[_CrewScope]

Keep a kickoff's policy decision with its caller, including async callers.

kyvvu_crewai.KyvvuCrewAIListener.setup_listeners

setup_listeners(crewai_event_bus: Any) -> None

Register all event handlers on the CrewAI event bus.

Called by BaseEventListener.__init__ during construction.

Parameters:

  • crewai_event_bus (Any) – The CrewAI global event bus instance.

kyvvu_crewai.KyvvuCrewAIListener.template_name

template_name: str = 'crewai'

kyvvu_crewai.KyvvuCrewAIListener.template_package

template_package: str = 'kyvvu_crewai.templates'

kyvvu_crewai.KyvvuCrewAIListener.tool_is_declared

tool_is_declared(name: str, declarations: frozenset[str]) -> bool

Check whether a declared tool has the normalized name.

kyvvu_crewai.__version__

CrewAI integration release version.