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 (step_type, scope, verb, properties) tuples. Rules are evaluated in order; first match wins.

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

rules:
  - name: model_call
    conditions:
      event_type: llm_invoke
    output:
      step_type: step.model
      scope: step
      verb: POST
      properties:
        model:
          provider: "{{ provider }}"
          name: "{{ model_name }}"

  - name: tool_call
    conditions:
      event_type: tool_execute
    output:
      step_type: step.resource
      scope: step
      verb: POST

  - name: fallback
    conditions: {}
    output:
      step_type: step.unknown
      scope: step

Template variables ({{ provider }}) are interpolated from the event data at match time.

Loading and testing the template

from kyvvu.templates import BehaviorTemplate

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

# Test matching
result = template.match({
    "event_type": "llm_invoke",
    "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 interface between your framework and the SDK:

from kyvvu.integrations import FrameworkAdapter

class MyFrameworkAdapter(FrameworkAdapter):
    def on_event(self, event_type: str, event_data: dict) -> None:
        """Called by your framework hook/callback."""
        # Construct event dict for template matching
        template_input = {
            "event_type": event_type,
            **event_data,
        }

        # Match template and evaluate
        behavior = self.match_and_evaluate(template_input)

        # If not blocked, the step was recorded automatically

Base class methods

Method	Purpose
match_and_evaluate(event_data)	Match template, evaluate policies, execute, record. Returns the Behavior. Raises KyvvuBlockedError on block.
start_task(**kwargs)	Begin a new task.
end_task(**kwargs)	End the current task.
error_task(**kwargs)	End the current task with an error.

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:

def test_llm_call_maps_to_step_model():
    template = BehaviorTemplate.from_path("my_framework.template.yaml")
    result = template.match({"event_type": "llm_invoke", "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_type": "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

from kyvvu import Kyvvu
from kyvvu.integrations import FrameworkAdapter

class CrewAIAdapter(FrameworkAdapter):
    def __init__(self, kv: Kyvvu):
        super().__init__(kv, template_name="crewai")

    def on_agent_action(self, agent_name: str, action: str, input_data: dict):
        self.match_and_evaluate({
            "event_type": "agent_action",
            "agent_name": agent_name,
            "action": action,
            **input_data,
        })

    def on_tool_use(self, tool_name: str, args: dict):
        self.match_and_evaluate({
            "event_type": "tool_execute",
            "tool_name": tool_name,
            **args,
        })

---

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