Query Data via SDKs

Langfuse is open-source and data tracked with Langfuse is open. Use the Python and JS/TS SDKs to query the same public APIs without writing raw HTTP requests.

Common use cases:

Query row-level observations for evaluation pipelines, few-shot examples, or fine-tuning datasets.
Query aggregate cost, usage, latency, volume, and score metrics for dashboards or billing workflows.
Programmatically create datasets.

If you are new to Langfuse, we recommend familiarizing yourself with the Langfuse data model.

Ingestion is asynchronous: SDK flush() only guarantees delivery to the API, not read visibility. New data is typically available for querying within 15-30 seconds of ingestion, though processing times may vary — under heavy or parallel load, visibility can lag longer (there is no fixed SLA). Reading a trace immediately after writing it may return a 404. See Handling ingestion lag for the recommended retry pattern, and visit status.langfuse.com if you encounter persistent issues.

SDKs

Via the SDKs for Python and JS/TS you can easily query the API without having to write the HTTP requests yourself.

The api namespace is auto-generated from the Public API (OpenAPI). Method names mirror REST resources and support filters and pagination.

From Python SDK v4 and JS/TS SDK v5 onward, the high-performance v2 data APIs are the defaults:

api.observations (formerly api.observations_v_2 / api.observationsV2)
api.scores (formerly api.score_v_2 / api.scoreV2)
api.metrics (formerly api.metrics_v_2 / api.metricsV2)

The old v2 aliases were removed in Python SDK v4 and JS/TS SDK v5.

The older trace, observation, and metrics read APIs remain available, but they are not recommended as the default for new data extraction workflows because they are less performant at scale. See Observations API v2 for row-level data and Metrics API v2 for aggregates.

pip install langfuse

from langfuse import get_client
langfuse = get_client()  # uses environment variables to authenticate

Observations

observations = langfuse.api.observations.get_many(
    trace_id="abcdef1234",
    type="GENERATION",
    limit=100,
    fields="core,basic,usage"
)

Use trace_id to retrieve the observations that belong to a single trace. Use parent_observation_id in the response to reconstruct an observation tree when needed.

Traces: `list` returns lightweight views

langfuse.api.trace.list(...) returns lightweight trace views: the observations and scores fields contain only IDs (strings), not full objects. Trace-level aggregates (total_cost, latency) are included, but per-observation usage and cost details are not. To get full observation objects, call langfuse.api.trace.get(trace_id) per trace (or prefer observations.get_many(trace_id=...) above).

List-then-get pagination example for usage/cost aggregation:

from collections import defaultdict
from datetime import datetime, timedelta, timezone

usage_by_model = defaultdict(int)
from_timestamp = datetime.now(timezone.utc) - timedelta(days=1)
page = 1

while True:
    traces = langfuse.api.trace.list(
        from_timestamp=from_timestamp,
        page=page,
        limit=50,
    )
    for trace_view in traces.data:
        # trace_view.observations is a list of ID strings only
        full_trace = langfuse.api.trace.get(trace_view.id)
        for obs in full_trace.observations:  # full objects incl. usage/cost
            if obs.usage_details:
                model = obs.model or "unknown"
                usage_by_model[model] += obs.usage_details.get("total", 0)
    if page >= traces.meta.total_pages:
        break
    page += 1

If you only need trace-level totals, trace.list alone is enough (trace_view.total_cost, trace_view.latency) — no per-trace get required. For large-scale aggregation, prefer the Metrics API v2 over paginating row-level data.

Metrics

For the full query schema, supported dimensions, filters, and examples, see the Metrics API v2 documentation.

query = """
{
  "view": "observations",
  "metrics": [{"measure": "totalCost", "aggregation": "sum"}],
  "dimensions": [{"field": "providedModelName"}],
  "filters": [],
  "fromTimestamp": "2025-05-01T00:00:00Z",
  "toTimestamp": "2025-05-13T00:00:00Z"
}
"""

metrics = langfuse.api.metrics.get(query = query)

Other resources

Sessions:

sessions = langfuse.api.sessions.list(limit=50)

Scores:

scores = langfuse.api.scores.get_many(score_ids = "ScoreId")

Prompts:

Please refer to the prompt management documentation on fetching prompts.

Datasets:

# Namespaces:
# - langfuse.api.datasets.*
# - langfuse.api.dataset_items.*
# - langfuse.api.dataset_run_items.*

Async equivalents

# All endpoints are also available as async under `async_api`:
observations = await langfuse.async_api.observations.get_many(
    trace_id="abcdef1234",
    limit=100,
    fields="core,basic,usage",
)
metrics = await langfuse.async_api.metrics.get(query = query)

For row-level observation filters, field selection, and cursor pagination, see the Observations API v2 documentation.

The methods on the langfuse.api are auto-generated from the API reference and cover all entities. You can explore more entities via Intellisense

npm install @langfuse/client

import { LangfuseClient } from "@langfuse/client";

const langfuse = new LangfuseClient();

Observations

const observations = await langfuse.api.observations.getMany({
  traceId: "abcdef1234",
  type: "GENERATION",
  limit: 100,
  fields: "core,basic,usage",
});

Use traceId to retrieve the observations that belong to a single trace. Use parentObservationId in the response to reconstruct an observation tree when needed.

For row-level observation filters, field selection, and cursor pagination, see the Observations API v2 documentation.

Metrics

For the full query schema, supported dimensions, filters, and examples, see the Metrics API v2 documentation.

const query = {
  view: "observations",
  metrics: [{ measure: "totalCost", aggregation: "sum" }],
  dimensions: [{ field: "providedModelName" }],
  filters: [],
  fromTimestamp: "2025-05-01T00:00:00Z",
  toTimestamp: "2025-05-13T00:00:00Z",
};

const metrics = await langfuse.api.metrics.get({
  query: JSON.stringify(query),
});

Other resources

Sessions:

const sessions = await langfuse.api.sessions.list({ limit: 50 });

Scores:

const scores = await langfuse.api.scores.getMany();

Prompts:

Please refer to the prompt management documentation on fetching prompts.

Datasets:

// Namespaces:
// - langfuse.api.datasets.*
// - langfuse.api.datasetItems.*
// - langfuse.api.datasetRunItems.*

Explore more entities via Intellisense on langfuse.api.

Handling ingestion lag (get-after-write)

Trace ingestion is asynchronous. flush() in the SDKs guarantees that data was delivered to the Langfuse API, not that it is readable yet. Reading right after writing — e.g. trace.get(trace_id) after a flush, fetching a score you just created, or fetching dataset run items right after an experiment — may raise a 404 (langfuse.api.NotFoundError in Python) or return incomplete data until processing completes. Typical visibility is within 15-30 seconds; under heavy or parallel load it can take longer, and there is no fixed SLA.

Instead of a fixed time.sleep(), use bounded retries with backoff:

import time

from langfuse import get_client
from langfuse.api import NotFoundError

langfuse = get_client()


def get_trace_with_retry(trace_id: str, *, timeout: float = 60.0):
    """Fetch a trace, retrying on 404 until ingestion completes."""
    deadline = time.monotonic() + timeout
    delay = 1.0
    while True:
        try:
            return langfuse.api.trace.get(trace_id)
        except NotFoundError:
            if time.monotonic() >= deadline:
                raise
            time.sleep(delay)
            delay = min(delay * 2, 10.0)  # exponential backoff, capped


langfuse.flush()  # ensure delivery before polling
trace = get_trace_with_retry(trace_id, timeout=60)

Note: a successful trace.get means the trace record exists — observations and scores of that trace may still be arriving. If you depend on completeness (e.g. counting observations or summing costs), additionally check that the expected observations are present before proceeding.

import { LangfuseClient } from "@langfuse/client";

const langfuse = new LangfuseClient();

async function getTraceWithRetry(traceId: string, timeoutMs = 60_000) {
  const deadline = Date.now() + timeoutMs;
  let delay = 1_000;
  while (true) {
    try {
      return await langfuse.api.trace.get(traceId);
    } catch (err: any) {
      const is404 = err?.statusCode === 404;
      if (!is404 || Date.now() >= deadline) throw err;
      await new Promise((r) => setTimeout(r, delay));
      delay = Math.min(delay * 2, 10_000); // exponential backoff, capped
    }
  }
}

The same pattern applies to any read-after-write on ingested entities: traces, observations, scores, and dataset run items. Entities created via synchronous endpoints (prompts, datasets, dataset items) are readable immediately and do not need retries.

To move existing trace or observation reads to v2, see Observations API v2.
To move existing metrics queries to v2, see Metrics API v2.
For large-scale data exports (e.g., all traces for fine-tuning or analytics), consider using the Blob Storage Export to automatically sync data to S3, GCS, or Azure on a schedule instead of paginating through the API.
To manually export filtered data from the Langfuse UI, see Export from UI.

Was this page helpful?

On this page