Langfuse Python SDK

Use get_client() to initialize the Langfuse client. You can then use start_as_current_observation as a context manager to create nested traces.

• Use as_type="span" for general processing steps.
• Use as_type="generation" for LLM calls.
• Use .update(output=...) to record the result of an operation.
• Observations are automatically closed when exiting the with block.
• Call .flush() in short-lived applications to ensure all events are sent before the process exits.

# env: LANGFUSE_PUBLIC_KEY, LANGFUSE_SECRET_KEY, LANGFUSE_BASE_URL

from langfuse import get_client

langfuse = get_client()

# Create a span using a context manager
with langfuse.start_as_current_observation(as_type="span", name="process-request") as span:
    # Your processing logic here
    span.update(output="Processing complete")

    # Create a nested generation for an LLM call
    with langfuse.start_as_current_observation(as_type="generation", name="llm-response", model="gpt-5.6") as generation:
        # Your LLM call logic here
        generation.update(output="Generated response")

# All spans are automatically closed when exiting their context blocks

# Flush events in short-lived applications
langfuse.flush()

The metrics method requires a JSON string query with the following schema:

Aggregation functions: sum, avg, count, max, min, p50, p75, p90, p95, p99, histogram.

{
  "view": "observations",
  "dimensions": [
    { "field": "type" }
  ],
  "metrics": [
    { "measure": "latency", "aggregation": "avg" }
  ],
  "fromTimestamp": "2023-01-01T00:00:00Z",
  "toTimestamp": "2023-01-02T00:00:00Z"
}

The Langfuse client provides access to the full Langfuse REST API through the .api property (synchronous) and .async_api property (asynchronous). Use these to read or manage data on the Langfuse server (traces, observations, scores, datasets, prompts, etc.).

Important Semantics:

• Ingestion is asynchronous: langfuse.flush() only guarantees delivery to the API, not immediate read visibility. A trace might not be visible via api.trace.get(trace_id) for 15-30 seconds after ingestion. Use retries with a deadline instead of fixed sleeps.
• List vs. Get: List endpoints (e.g., api.trace.list(...)) return lightweight views where observations and scores are lists of IDs. To get full details, use api.trace.get(trace_id) or api.observations.get_many(trace_id=...).
• Prefer v2 Data APIs: Since SDK v4, api.observations and api.metrics map to high-performance /api/public/v2/... endpoints and are the recommended read path. Avoid api.legacy endpoints for new workflows.
• Aggregation: For large-scale aggregation (usage/cost by model, user, etc.), use the v2 Metrics API (api.metrics.metrics(...)) instead of paginating through row-level data.

While get_many supports individual query parameters (like user_id or type), you can use the filter parameter for complex queries. The filter parameter accepts a JSON string containing an array of filter conditions. When provided, the filter parameter takes precedence over individual query parameter filters.

Filter Condition Structure

Each condition requires:

• type: One of datetime, string, number, stringOptions, categoryOptions, arrayOptions, stringObject, numberObject, boolean, null.
• column: The field to filter on (e.g., latency, metadata, input).
• operator: Based on type (e.g., for string: =, contains, matches; for datetime: >, <=).
• value: The value to compare against.
• key: Required for stringObject, numberObject, and categoryOptions when filtering nested fields like metadata.

[
  {
    "type": "string",
    "column": "type",
    "operator": "=",
    "value": "GENERATION"
  },
  {
    "type": "number",
    "column": "latency",
    "operator": ">=",
    "value": 2.5
  },
  {
    "type": "stringObject",
    "column": "metadata",
    "key": "environment",
    "operator": "=",
    "value": "production"
  }
]