Writing a New Integration

What you'll learn: How to integrate Kyvvu with a framework that isn't covered by the built-in decorator or LangChain handler, by authoring a YAML template and subclassing FrameworkAdapter.

---

Overview

Integrating a new framework requires two things:

1. A YAML template — defines how the framework's events map to Kyvvu's atomic Behavior vocabulary.

2. A FrameworkAdapter subclass — captures the framework's events and passes them through the template system.

Step 1: Author a YAML template

For a comprehensive guide on template authoring -- match semantics, property extraction, and a full worked example -- see the Template Authoring Guide.

The template maps framework events to behaviour fields (step_type, verb, step_name, properties). Rules are evaluated in order; first match wins. Each rule has a match block (conditions) and a behavior block (the emitted behaviour):

name: my-framework-template
version: "1.0"

rules:
  - id: model_call
    match:
      event: "llm_invoke"
    behavior:
      step_type: "step.model"
      verb: "POST"
      step_name: "{{ name }}"
      properties:
        model:
          provider: "{{ provider }}"
          name: "{{ model_name }}"

  - id: tool_call
    match:
      event: "tool_execute"
    behavior:
      step_type: "step.resource"
      verb: "POST"
      step_name: "{{ name }}"

  - id: fallback
    match: {}
    behavior:
      step_type: "step.unknown"

Match conditions support event (exact match) and name_pattern (regex on the name context key). Template variables ({{ provider }}) are interpolated from the event context at match time. See Writing a Custom Template for the full match/behavior field reference.

Loading and testing the template

from kyvvu_engine.templates import BehaviorTemplate

template = BehaviorTemplate.from_path("my_framework.template.yaml")

# Test matching
result = template.match({
    "event": "llm_invoke",
    "name": "chat",
    "provider": "openai",
    "model_name": "gpt-4o",
})
assert result["step_type"] == "step.model"
assert result["properties"]["model"]["provider"] == "openai"

Step 2: Subclass FrameworkAdapter

The FrameworkAdapter base class provides the machinery between your framework and the SDK. Set the template_name class attribute to the name of your built-in template, then write callback methods that build a context dict and delegate to the base class. The constructor takes an already-constructed Kyvvu instance (the adapter does not manage identity or registration) plus an optional template override:

from kyvvu.integrations import FrameworkAdapter

class MyFrameworkAdapter(FrameworkAdapter):
    template_name = "myframework"  # resolves myframework.template.yaml

    def on_tool_event(self, tool_name: str, args: dict, run_id: str) -> None:
        """Called by your framework hook/callback."""
        context = {
            "event": "tool_execute",
            "name": tool_name,
            "task_id": self.kv.start_task() if ... else current_task_id,
        }
        # Two-phase emission: evaluate (pre-execution) then record (post).
        self._emit_step(context, input_data=args, step_run_id=run_id)
        # ... framework executes the tool ...
        self._record_step(run_id, output={"status": "success", "result": result})

Base class methods

Method	Purpose
_emit_step(context, input_data, step_run_id)	Match the template, evaluate policies for the intended step, and buffer it. Raises KyvvuBlockedError on block (records the blocked step and flushes the task).
_record_step(step_run_id, output=...)	Complete a step previously started by _emit_step and record it into history.
_emit_task_start(context, task_id=None)	Emit task.start and return the task_id.
_emit_task_end(context, task_id=None)	Emit task.end and flush the task.
_emit_task_error(context, error, task_id=None)	Emit task.error and flush the task.
_resolve_root(run_id, parent_run_id)	Resolve the root task_id for a nested callback.
_safe_end_task(task_id, emit_fn)	Idempotent, crash-safe task flush (re-raises KyvvuBlockedError).

Subclasses may override the hooks _build_properties(matched, context, step_run_id) (inject framework-specific metadata before policy evaluation) and _on_block(task_id, exc) (circuit-breaker behaviour on block — CrewAI uses this because its event bus swallows exceptions).

Integration with the framework

How you hook into the framework depends on the framework. Common patterns:

• Callback/hook system — register your adapter as a callback (like LangChain's BaseCallbackHandler).

• Middleware — insert the adapter as middleware in the framework's request pipeline.

• Monkey-patching — wrap the framework's core methods (least preferred).

Step 3: Test the mapping

Write tests that verify your template produces the correct Behaviors for each framework event:

from kyvvu_engine.templates import BehaviorTemplate

def test_llm_call_maps_to_step_model():
    template = BehaviorTemplate.from_path("my_framework.template.yaml")
    result = template.match({"event": "llm_invoke", "name": "chat", "model_name": "gpt-4o"})
    assert result["step_type"] == "step.model"
    assert result["verb"] == "POST"

def test_unknown_event_falls_back():
    template = BehaviorTemplate.from_path("my_framework.template.yaml")
    result = template.match({"event": "something_unexpected"})
    assert result["step_type"] == "step.unknown"

Test each event type your framework emits, including edge cases (nested events, error events, events with missing fields).

Example: complete adapter

template_name is a class attribute; the constructor takes the Kyvvu instance. Each handler builds a context dict and drives the two-phase _emit_step / _record_step cycle (or the _emit_task_* lifecycle helpers):

import uuid
from kyvvu import Kyvvu
from kyvvu.integrations import FrameworkAdapter

class MyFrameworkAdapter(FrameworkAdapter):
    template_name = "myframework"  # resolves myframework.template.yaml

    def on_kickoff(self) -> None:
        self._task_id = self._emit_task_start({"event": "kickoff", "name": "run"})

    def on_tool_use(self, tool_name: str, args: dict) -> None:
        run_id = str(uuid.uuid4())
        context = {"event": "tool_execute", "name": tool_name, "task_id": self._task_id}
        self._emit_step(context, input_data=args, step_run_id=run_id)
        result = ...  # framework executes the tool
        self._record_step(run_id, output={"status": "success", "result": result})

    def on_finish(self) -> None:
        self._safe_end_task(
            self._task_id,
            lambda: self._emit_task_end({"event": "finish", "name": "run"}, task_id=self._task_id),
        )

The built-in CrewAI integration (KyvvuCrewAIListener, kyvvu.integrations.crewai) is a worked reference implementation of this pattern against the CrewAI event bus.

---

Next steps

• Templates — deep-merge semantics and template loading options

• Atomic Behaviours — the 12 behaviour types to map to

• REST API (Non-Python) — for non-Python agents, use the HTTP server instead