No importer for your format

No built-in importer for your format? An importer is one Python callable: scaffold it, test it offline, register it, and your export imports like any other.

The built-in importers cover Langfuse, LangSmith, Braintrust, Logfire, Arize Phoenix, and the Kitaru JSONL contract. Any other trace store, or a homegrown logging format, comes in through a custom importer. An importer is small by design: one callable that parses your export bytes into sessions, usually about a page of Python.

There are two ways to get one, and the fast way is to not write it yourself: the kitaru-importer-builder agent skill turns a representative export into a locally validated importer. It keeps the mapping from source evidence to normalized sessions explicit, so you can see what is preserved, approximated, or unavailable, and it finishes on your machine until you approve registration.

The contract

from collections.abc import Iterator
from typing import Any

from kitaru.task.importer import ImportFailure, ParsedNode, ParsedSession


def parse(
    payload: bytes, params: dict[str, Any]
) -> Iterator[ParsedSession | ImportFailure]:
    for line_number, line in enumerate(payload.splitlines(), start=1):
        try:
            record = decode_my_format(line)
        except ValueError as error:
            yield ImportFailure(line=line_number, error=str(error))
            continue
        yield ParsedSession(
            status="completed",
            name=record.title,
            inputs=record.question,
            outputs=record.answer,
            error=None,
            started_at=record.started_at,
            ended_at=record.ended_at,
            external_id=record.trace_id,
            metadata={},
            nodes=[
                ParsedNode(
                    node_type="llm_call",
                    name="model",
                    status="completed",
                    inputs=record.prompt,
                    outputs=record.completion,
                ),
            ],
        )

Yield lazily; the import consumes one item at a time, so payload size is bounded by disk, not memory. Yield an ImportFailure for a bad record and the import counts it and moves on. Only a crash of the parser itself fails the task, with partial stats preserved. The full field reference for ParsedSession and ParsedNode is the portable session contract. parse may be a regular or an async generator.

Set a stable external_id from your source system: together with the importer's provider name it is the dedup key, so re-importing an overlapping export skips what is already stored instead of duplicating it.

Fetch traces from your own API instead of a file

A custom importer can accept an API import too, the same way the built-in provider importers do. Instead of a bare parse function, register an importer object as the entrypoint. It exposes parse and fetch, each of which may be a regular or an async generator:

from collections.abc import AsyncIterator, Iterator
from typing import Any


class MyImporter:
    def parse(
        self, payload: bytes, params: dict[str, Any]
    ) -> Iterator[ParsedSession]: ...

    async def fetch(self, query: dict[str, Any]) -> AsyncIterator[bytes]:
        for trace_id in select_traces(query):
            yield fetch_trace_bytes(trace_id)


importer = MyImporter()

Register it with --entrypoint importer for a script source, or my_importer:importer for a package source. An entrypoint that is a plain callable stays upload-only.

A package source declares what fetch needs under a api extra in its pyproject.toml, and the worker installs my-importer[api] for an API import and the bare package for an upload. A script source lists its dependencies inline as for any script plugin, so they are installed for both.

fetch receives the import's --query (or source.query on the request) and yields parser payloads. The server validates the shared keys (trace_ids, since, until, concurrency) as ImportQuery from kitaru.api_models.v1.imports before the import is created, and passes provider-specific keys such as project_id through untouched, so fetch receives the full merged dict. Each yielded payload runs through parse with the import's params and is ingested before the next payload is pulled, exactly like a file upload would. A session that a later payload yields again gets that payload's nodes added, but keeps the name, status, inputs, outputs, timestamps, and metadata of the payload that created it, so keep every trace of a session in one payload when parse derives those from all of its traces. The built-in importers resolve the session key from the provider's listing call and hold a session until every trace of it is fetched, yielding complete sessions in batches, oldest first, so imported sessions appear while the import is still running. Their bounded concurrency relies on fetch being an async generator. A custom join_on is not visible to fetch, so an API import with a custom join key only groups traces that also share the provider's default session key. The task closes the fetch generator when the import stops early, for example at --max-sessions, so put request cleanup in a finally block. Raise from fetch to end the import task with the failure recorded in the import stats. An API import against an importer without fetch fails the same way, so kitaru session import --wait reports it in the import stats.

Scaffold, test offline, register

kitaru importer scaffold my-format          # writes my_format_importer.py
kitaru importer test my_format_importer.py \
  --entrypoint parse --payload sample-export.jsonl
kitaru importer register my-format \
  --script my_format_importer.py --entrypoint parse --provider my-format

A script importer may declare dependencies as PEP 723 inline metadata (a # /// script block); the worker builds it an isolated environment. An importer that outgrows one file ships as a package instead: --package "my-importer==1.0.0" with --entrypoint "my_importer:parse". Importers are versioned like evaluators and agents; imports name the importer and pin to its latest version unless you pass one.

If your format's fetch reads provider credentials from the environment, declare them as a --connection-schema FILE on register, a JSON Schema whose properties are the environment variable names, with writeOnly: true marking a property as secret. kitaru connection create --importer my-format then prompts for those properties instead of requiring --set/--set-secret for keys you'd otherwise have to remember. See Provider connections.

Once registered, your format imports exactly like the built-in ones:

kitaru session import my-export.jsonl \
  --importer my-format@latest \
  --agent support-agent@latest --wait

The shipped importers are the reference implementations: plugins/packages/jsonl-importer is the smallest at under 80 lines, and the Langfuse one shows real normalization with turn grouping and warnings.

Imported payloads contain whatever your traces contain: prompts, customer data, tool results. They are stored on your self-hosted server and parsed on your workers, but access and retention are yours to govern.

Next

Evaluate the history you imported with Write an evaluator, then freeze the sessions that matter into a cohort and put a change to the test with Build a regression suite from production.