> For the complete documentation index, see [llms.txt](https://docs.kyvvu.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.kyvvu.com/integrations/custom.md).

# 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](/integrations/authoring.md).

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

```yaml
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](/integrations/custom-template.md) for the full match/behavior field reference.

### Loading and testing the template

```python
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:

```python
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:

```python
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):

```python
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](/core-concepts/templates.md) — deep-merge semantics and template loading options
* [Atomic Behaviours](/core-concepts/behaviours.md) — the 12 behaviour types to map to
* [REST API (Non-Python)](/integrations/rest-api.md) — for non-Python agents, use the HTTP server instead