Langfuse

Langfuse JSONL imports end to end: the accepted export, dedup semantics, and how imports execute on your worker.

Import your traces covers the shortest path: one kitaru session import against the built-in Langfuse importer. This guide is the full contract: what the importer understands, how re-runs dedup, and how to write an importer for any other format.

How an import executes

An import is a job with one importer task. You upload the export as a blob; a worker claims the task, materializes the importer's code and your payload, and runs the parse in your environment; the server never parses your data. Each parsed trace becomes one session with origin: imported, its observations ingested as nodes in batches.

The CLI wraps the upload and the job in one command:

kitaru session import langfuse-export.jsonl \
  --importer kitaru/langfuse@latest \
  --agent support-agent@latest \
  --params '{"source_instance": "my-langfuse-project"}' \
  --media-type application/x-ndjson \
  --tag imported-baseline --wait

--tag labels the created sessions once the import completes (so it requires --wait); downstream commands select on it. On the Python client the same import is explicit:

from kitaru.api_models.v1.imports import ImportCreateRequest

job = await client.imports.create(
    ImportCreateRequest(
        importer="langfuse",  # importer name in the registry
        agent_id=AGENT_ID,  # sessions land under this agent
        agent_version_id=None,  # optional: stamp a version on them
        payload_blob_id=blob.id,
        params={"source_instance": "my-langfuse-project"},
    )
)

Set agent_version_id when you know which code produced the traces; it's what lets a later replay default to the right version. The task's result carries the stats: sessions created, skipped, failed, with up to 20 failure samples (line number, external id, error).

The built-in importers

Kitaru ships provider importers as default plugins, registered at server startup under the kitaru/ namespace, so --importer kitaru/langfuse@latest always resolves. They run on your worker like any other importer; there is nothing to write. See Import your traces for the current built-in list.

The Langfuse importer parses Langfuse JSONL exports, with uploads capped by the server's configurable blob limit, and understands three record shapes: trace, observation, and raw ingestion_event lines. Traces map to sessions; observations map to nodes with their parent relationships, timings, model names, token usage, and cost preserved. params:

Param	Meaning
source_instance	Stable source project identity. Required for file uploads without an embedded project ID, unless project_id is supplied instead. Takes precedence over project_id and embedded identity.
project_id	Provider-native alias for source_instance, used when source_instance is absent or empty.
infer_tool_call_links	Optional boolean, default true. The importer matches tool-call ids emitted by a generation with gen_ai.tool.call.id on tool observations, nests each unambiguous tool call under the requesting generation, and keeps its original Langfuse parent as a link of kind source_parent. Unmatched or ambiguous ids remain unchanged. Set this to false to keep only the source observation hierarchy.

For UI and events exports without project IDs, pass --params '{"source_instance":"my-langfuse-project"}' or --params '{"project_id":"my-langfuse-project"}'. SDK and REST callers supply the same parameters on import creation. Keep the value stable across exports of the same project; filenames do not determine identity. If earlier imports used a filename stem as their identity, supply that same value explicitly to preserve deduplication.

Identity values are trimmed strings. Conflicting embedded project IDs fail the affected session even with an explicit override. See Import your traces for the shared identity rules and guidance for existing imports.

Fetch traces from the Langfuse API

Skip the export and upload, and let the import task fetch traces from Langfuse directly:

kitaru session import \
  --importer kitaru/langfuse@latest \
  --agent support-agent@latest \
  --since 7d \
  --tag imported-baseline --wait

Omitting FILE and setting --since selects an API import: the worker calls the Langfuse API instead of parsing an uploaded payload. --since and --until accept an ISO 8601 timestamp or a relative duration (7d, 12h, 30m). --trace-id (repeatable) fetches exactly those trace ids instead of a time window. The same selection is a query object on the SDK and REST request:

Query key	Meaning
trace_ids	Langfuse trace ids to fetch. When present, exactly those traces are fetched and the time window is ignored.
since	Timezone-aware ISO 8601 datetime, lower bound of trace start time. Required when trace_ids is absent.
until	Timezone-aware ISO 8601 datetime, upper bound of trace start time. Defaults to now.
concurrency	Traces fetched at once. Defaults to 4.

The worker installs the package's api extra for an API import, which carries the provider client. A connection you name with --connection, or the provider's default connection, supplies LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, and LANGFUSE_BASE_URL (or the older LANGFUSE_HOST for a self-hosted instance). Without either, the worker's own environment does, and only a worker started with --selector kitaru/requires-credentials=langfuse claims the task. Each fetched trace is parsed the same way an uploaded export would be, so the params table above, and the dedup rules below, apply the same way.

Dedup: one session per (imported_from, external_id) per agent

Every imported session records imported_from (langfuse) and an external_id combining the selected source identity with the source session ID. This pair is unique per destination agent, so re-importing an overlapping export with the same identity skips what's already stored. The stats report it as skipped, not as an error. Skipped sessions are not updated with new nodes.

No importer for your format?

The importer contract is deliberately small, about a page of Python, and the shipped Langfuse importer is a reference implementation of it. See No importer for your format to scaffold, test, and register your own.

After the import

Imported sessions are full Kitaru sessions: evaluate them with evaluators (backfilling your history is a single batch call), freeze them into cohorts, and replay them. Replay re-runs your code, which no trace export contains, so the agent's code must be registered as an agent version with a run command.