> For the complete documentation index, see [llms.txt](https://docs.zenml.io/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.zenml.io/kitaru/import-your-traces/custom-importer.md).

# 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](/kitaru/import-your-traces/import-langfuse-traces.md), [LangSmith](/kitaru/import-your-traces/import-langsmith-traces.md), [Braintrust](/kitaru/import-your-traces/import-braintrust-traces.md), [Logfire](/kitaru/import-your-traces/import-logfire-traces.md), [Arize Phoenix](/kitaru/import-your-traces/import-phoenix-traces.md), and the [Kitaru JSONL contract](/kitaru/import-your-traces/importing-sessions.md). 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](/kitaru/getting-started/setup.md) 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

```python
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](/kitaru/import-your-traces/importing-sessions.md). `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:

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

```bash
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](/kitaru/import-your-traces/provider-connections.md).

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

```bash
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.

{% hint style="warning" %}
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.
{% endhint %}

## Next

Evaluate the history you imported with [Write an evaluator](/kitaru/guides/write-an-evaluator.md), then freeze the sessions that matter into a cohort and put a change to the test with [Build a regression suite from production](/kitaru/guides/regression-suite.md).


---

# Agent Instructions
This documentation is published with GitBook. GitBook is the documentation platform designed so that both humans and AI agents can read, navigate, and reason over technical content effectively. Learn more at gitbook.com.

## Querying This Documentation
If you need additional information that is not directly available in this page, you can query the documentation dynamically by asking a question.

Perform an HTTP GET request on the current page URL with the `ask` query parameter, and the optional `goal` query parameter:

```
GET https://docs.zenml.io/kitaru/import-your-traces/custom-importer.md?ask=<question>&goal=<endgoal>
```

`ask` is the immediate question: it should be specific, self-contained, and written in natural language.
`goal` is optional and describes the broader end goal you are ultimately trying to accomplish on behalf of the user. GitBook uses it to tailor the answer towards what is most useful for that goal.

The response will contain a direct answer to the question and relevant excerpts and sources from the documentation.

Use this mechanism when the answer is not explicitly present in the current page, you need clarification or additional context, or you want to retrieve related documentation sections.