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¶

The built-in LangChain, LangGraph and CrewAI adapters classify tool callbacks from Kyvvu(read_only_tools=[...], write_tools=[...]), with POST as the default. A custom adapter with its own event names uses the verbs its template supplies.

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 (kind, 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: kickoff
    match:
      event: "kickoff"
    behavior:
      kind: "task.start"
      step_name: "{{ name }}"

  - id: finish
    match:
      event: "finish"
    behavior:
      kind: "task.end"
      step_name: "{{ name }}"

  - id: model_call
    match:
      event: "llm_invoke"
    behavior:
      kind: "step.model"            # model calls carry no verb
      step_name: "{{ name }}"
      properties:
        provider: "{{ provider | omit_empty }}"
        model: "{{ model_name }}"

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

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

Match conditions are an exact-equality check on any context key (event, kind, a dot-path such as serialized.name) and name_pattern, a case-insensitive re.search on the name context key (anchor with ^ and $). name_pattern is the one name key on every surface, and a context with no name never matches it. Every rule needs a unique id, and the template must contain a rule for each task boundary your adapter opens (here kickoff and finish). The template governs both the steps and those boundaries, and a template you supply replaces the built-in rather than extending it. 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 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["kind"] == "step.model"
assert result["properties"] == {"provider": "openai", "model": "gpt-4o"}

Step 2: Subclass FrameworkAdapter¶

The FrameworkAdapter base class provides the machinery between your framework and the Python client. 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) and a template.

Set template_name to your YAML filename without .template.yaml and template_package to the importable package that ships it. For example, myframework.template.yaml in myframework_templates/ uses template_name = "myframework" and template_package = "myframework_templates". The default constructor loads that package resource. A template passed to the constructor replaces it; KV_TEMPLATE_LOCATION replaces it when the default is used.

The examples below use an existing kv client and a framework object whose execute(tool_name, args) method performs the tool call. Substitute your framework's executor at that point, between evaluation and recording.

The adapter owns its task. It opens the task with emit_task_start and closes it with emit_task_end or emit_task_error. Never call kv.start_task() yourself inside a callback or around the adapter: if a task is already active in the context, emit_task_start raises KyvvuConfigError.

from kyvvu import Kyvvu
from kyvvu import FrameworkAdapter
from kyvvu import BehaviorTemplate

class MyFrameworkAdapter(FrameworkAdapter):
    template_name = "decorator"  # the fallback only; we always pass template=

    def on_tool_event(self, tool_name: str, args: dict, run_id: str) -> None:
        """Called by your framework hook/callback, inside a task the adapter opened."""
        context = {
            "event": "tool_execute",
            "name": tool_name,
            "task_id": self._task_id,  # set by your kickoff handler, see below
        }
        # Two-phase emission: evaluate (pre-execution) then record (post).
        self.emit_step(context, input_data=args, step_run_id=run_id)
        result = framework.execute(tool_name, args)  # your framework's executor
        self.record_step(run_id, output={"status": "success", "result": result})

template = BehaviorTemplate.from_path("my_framework.template.yaml")
adapter = MyFrameworkAdapter(kv, template=template)

Base class methods¶

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

The adapter exposes template, pending_steps, owned_tasks, and ended_tasks for subclass callbacks that need to inspect their own run state.

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 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["kind"] == "step.model"
    assert "verb" not in result  # step.model takes no verb

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

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

Example: complete adapter¶

The constructor takes the Kyvvu instance and the template you loaded. Each handler builds a context dict and drives the two-phase emit_step / record_step cycle (or the emit_task_* lifecycle helpers). The adapter opens and closes the task; the caller never calls kv.start_task():

import uuid
from kyvvu import Kyvvu, ActionKind
from kyvvu import FrameworkAdapter
from kyvvu import BehaviorTemplate

class MyFrameworkAdapter(FrameworkAdapter):
    template_name = "decorator"  # the fallback only; we always pass template=

    # The boundaries this adapter opens; the template is checked against
    # them when the adapter is built.
    boundary_contexts = {
        ActionKind.task_start: ({"event": "kickoff"},),
        ActionKind.task_end: ({"event": "finish"},),
    }

    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.execute(tool_name, args)  # your framework's executor
        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),
        )

adapter = MyFrameworkAdapter(
    kv, template=BehaviorTemplate.from_path("my_framework.template.yaml")
)

The built-in CrewAI integration (KyvvuCrewAIListener, kyvvu_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