Butterflow

Adapter Authoring Guide

Adapters bridge Butterflow to agent frameworks. This guide explains how to write a new adapter from scratch.

Base class

All adapters inherit from BaseAdapter:

from butterflow.adapters.base import BaseAdapter
from typing import Any

class MyAdapter(BaseAdapter):
    capability = 2
    adapter_id = "my_framework"

    def ingest(self, source: str | bytes, **kwargs: Any) -> list[dict[str, Any]]:
        ...

    def is_available(self) -> bool:
        ...

Capability levels

Level Name Required methods
1 Ingest only ingest(), is_available()
2 Execute + execute()
3 Normalize + normalize_tool_call(), normalize_state(), normalize_artifact()
4 Token fingerprints + token_fingerprint()
5 Repair hints + repair_hints()

Key methods

detect()

A classmethod that returns True when the target framework is present:

@classmethod
def detect(cls) -> bool:
    import importlib.util
    return importlib.util.find_spec("my_framework") is not None

ingest()

Convert a raw trace (file path, bytes, or string) into a list of normalized event dicts. Each dict must contain at minimum:

Example:

def ingest(self, source: str | bytes, **kwargs: Any) -> list[dict[str, Any]]:
    data = json.loads(source)
    return [
        {
            "event_type": "ToolCalled",
            "run_id": data["run_id"],
            "adapter_id": self.adapter_id,
            "tool_name": data["tool"],
            "args": data["args"],
        }
    ]

execute()

Run a flow dict and return a result dict with status and events:

def execute(self, flow: dict[str, Any], **kwargs: Any) -> dict[str, Any]:
    events = [...]  # drive the framework and collect events
    return {"status": "passed", "events": events}

is_available()

Return whether runtime dependencies are installed. For built-in adapters this can simply call self.detect().

Registration

Add your adapter to src/butterflow/registry/adapters.toml:

[adapters.my_framework]
name = "my_framework"
package = "butterflow-my-framework"
pip_extras = "butterflow[my-framework]"
capability = 2
detection_signals = ["my_framework", "MyAgent"]
description = "My custom framework adapter."
docs_url = ""

Testing

Unit-test ingestion with synthetic payloads:

def test_ingest_tool_called():
    adapter = MyAdapter()
    events = adapter.ingest('{"tool": "search", "args": {"q": "x"}}')
    assert events[0]["event_type"] == "ToolCalled"