OpenObserve Python SDK

A Python SDK for OpenObserve telemetry and evaluations.

Features

• Easy Integration – Minimal setup with automatic instrumentation for popular libraries

• Multi-Signal Support – Capture logs, metrics, and traces simultaneously

• Flexible Protocol – Choose between HTTP/Protobuf (default) or gRPC

• Agent Identity – Stamp GenAI agent identity on trace spans

• Experiments – Evaluate your own code against a dataset and gate CI on regressions

• Lightweight – Minimal dependencies, designed for production use

• OpenTelemetry Native – Built on OpenTelemetry standards for compatibility

Prompt Management

Resolve text prompts from an OpenObserve prompt registry with ETag-based refresh and immutable in-memory snapshots. This API is independent of OpenTelemetry and agent frameworks. See the prompts guide for setup, lifecycle, and error semantics.

Quick Start

Generate auth token:

echo -n "root@example.com:Complexpass#123" | base64
# Output: cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM=

Set environment variables:

# OpenObserve Configuration (Required)
export OPENOBSERVE_AUTH_TOKEN="Basic cm9vdEBleGFtcGxlLmNvbTpDb21wbGV4cGFzcyMxMjM="

# Optional OpenObserve settings (defaults shown)
export OPENOBSERVE_URL="http://localhost:5080"
export OPENOBSERVE_ORG="default"

# API keys for services you're using (optional, based on instrumentation)
export OPENAI_API_KEY="your-openai-key"
export ANTHROPIC_API_KEY="your-anthropic-key"

Install dependencies:

pip install openobserve-python-sdk openai opentelemetry-instrumentation-openai

Quick Example – OpenAI Instrumentation:

from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from openobserve import openobserve_init

# Initialize OpenObserve and instrument OpenAI
OpenAIInstrumentor().instrument()
openobserve_init()

from openai import OpenAI

# Use OpenAI as normal - traces are automatically captured
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hello!"}]
)
print(response.choices[0].message.content)

Quick Example – Anthropic Instrumentation:

from opentelemetry.instrumentation.anthropic import AnthropicInstrumentor
from openobserve import openobserve_init

# Initialize OpenObserve and instrument Anthropic
AnthropicInstrumentor().instrument()
openobserve_init()

from anthropic import Anthropic

# Use Claude as normal - traces are automatically captured
client = Anthropic()
response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.content[0].text)

Selecting Signals

By default, openobserve_init() initializes all signals (logs, metrics, traces). You can also initialize selectively:

# All signals (default)
openobserve_init()

# Specific signals only
openobserve_init(logs=True)
openobserve_init(metrics=True)
openobserve_init(traces=True)

# Combine signals
openobserve_init(logs=True, metrics=True)  # no traces

Note: For logs, you still need to bridge Python's standard logging module:

import logging
from opentelemetry.sdk._logs import LoggingHandler

openobserve_init(logs=True)
handler = LoggingHandler()
logging.getLogger().addHandler(handler)

Environment Variables

Variable	Required	Description
OPENOBSERVE_URL	No	OpenObserve base URL (default: "http://localhost:5080")
OPENOBSERVE_ORG	No	Organization name (default: "default")
OPENOBSERVE_AUTH_TOKEN	✅	Authorization token (Format: "Basic ")
OPENOBSERVE_TIMEOUT	No	Request timeout in seconds (default: 30)
OPENOBSERVE_ENABLED	No	Enable/disable telemetry(default: "true")
OPENOBSERVE_PROTOCOL	No	Protocol: "grpc" or "http/protobuf" (default: "http/protobuf")
OPENOBSERVE_TRACES_STREAM_NAME	No	Stream name for traces (default: "default")
OPENOBSERVE_LOGS_STREAM_NAME	No	Stream name for logs (default: "default")
OPENOBSERVE_AGENT_ID	No	GenAI agent ID to stamp on trace spans
OPENOBSERVE_AGENT_NAME	No	GenAI agent name to stamp on trace spans

Agent Identity

Use agent_id and/or agent_name to identify the GenAI agent that emitted trace spans:

from openobserve import openobserve_agent, openobserve_init

# Static identity for all trace spans from this process.
openobserve_init(agent_id="support-agent", agent_name="Support Agent")

# Request-scoped identity overrides static identity and propagates via OTel baggage.
with openobserve_agent(agent_name="Triage Agent"):
    run_agent_workflow()

The SDK stamps identity as span attributes (gen_ai.agent.id, gen_ai.agent.name). Span attributes are the preferred path for OpenObserve agent attribution, especially when a process can handle multiple agents or request-scoped agent identity.

For a single-agent process, you may also set the agent name as an OpenTelemetry resource attribute:

from openobserve import openobserve_init

openobserve_init(
    resource_attributes={
        "service.name": "support-agent-worker",
        "gen_ai.agent.name": "Support Agent",
    },
)

Resource-level gen_ai.agent.name is attached through the OpenTelemetry Resource. OpenObserve can use it as a fallback for LLM span agent identity, but span attributes take precedence. Use this only when the process has one static agent identity. For request-scoped or multi-agent processes, prefer agent_name= or openobserve_agent(...).

If you already manage OpenTelemetry providers yourself, see Native OpenTelemetry Agent Identity for equivalent native SDK patterns.

Protocol Configuration Notes

HTTP/Protobuf (default)

• Uses HTTP with Protocol Buffers encoding.
• Works with both HTTP and HTTPS endpoints.
• Organization is specified in the URL path: /api/{org}/v1/{signal}, where {signal} is traces, logs, or metrics.
• Automatically adds the stream-name header from OPENOBSERVE_TRACES_STREAM_NAME for traces and OPENOBSERVE_LOGS_STREAM_NAME for logs.
• Standard HTTP header handling (preserves case).

gRPC

• Requires the optional gRPC extra: pip install openobserve-python-sdk[grpc].
• Uses gRPC protocol with automatic configuration:
  • Organization is passed as a header (not in the URL).
  • Automatically adds required headers:
    • organization: Set to OPENOBSERVE_ORG.
    • stream-name: Set to OPENOBSERVE_TRACES_STREAM_NAME for traces and OPENOBSERVE_LOGS_STREAM_NAME for logs.
  • Headers are normalized to lowercase per gRPC specification.
  • TLS is automatically configured based on URL scheme:
    • http:// URLs use insecure (non-TLS) connections.
    • https:// URLs use secure (TLS) connections.

Installation

Choose your preferred installation method:

# From PyPI (recommended)
pip install openobserve-python-sdk

# With gRPC transport support (needed for protocol="grpc")
pip install "openobserve-python-sdk[grpc]"

# From source (development)
pip install -e .

# Using requirements.txt
pip install -r requirements.txt

HTTP/Protobuf is the default protocol. For gRPC, install openobserve-python-sdk[grpc] and set protocol="grpc" or OPENOBSERVE_PROTOCOL=grpc. The core package does not depend on grpcio.

Supported Instruments

The SDK works with OpenTelemetry instrumentation packages:

• OpenAI – Use with opentelemetry-instrumentation-openai for API call traces
• Anthropic – Use with opentelemetry-instrumentation-anthropic for Claude API traces
• LangChain – Use with opentelemetry-instrumentation-langchain for LLM chain tracing
• Standard Python Logging – Built-in support via LoggingHandler
• Metrics – OpenTelemetry counters, histograms, and up/down counters

Experiments

Evaluate your own code against a dataset, score the results, and fail CI when quality regresses. Experiments reuse the same environment variables as telemetry export, so an instrumented process needs no extra configuration.

Prepare the data

A run always anchors to a dataset that already exists — there is no data= parameter. A run that carried its own inline data would be comparable to no other run, which is the whole point of having one.

from openobserve import datasets, score_configs

datasets.upsert(
    "rag-qa-golden",
    items=[
        {
            "logical_id": "case-42",
            "input": {"question": "refund window?"},
            "expected_output": "30 days",
        },
    ],
)

score_configs.ensure(
    "exact_match",
    type="numeric",
    min=0,
    max=1,
    healthy_threshold={"direction": "gte", "value": 1.0},
)

upsert is safe to repeat: identical content appends no revision. Updating an existing logical_id requires the if_row_id you read, so a concurrent edit is a conflict rather than a silent overwrite.

ensure is safe to call on every run: identical parameters change nothing, and only a changed range, category set, or health policy appends a version.

Run it

from openobserve import experiment, scorer


@scorer(config="exact_match")
def exact_match(output, expected_output):
    return 1.0 if output.strip() == expected_output.strip() else 0.0


def my_task(input, context):
    # context carries row_id, trial_index, and the case's metadata
    return my_pipeline(input["question"])


result = experiment.run(
    "prompt-v3",
    dataset="rag-qa-golden",  # or "rag-qa-golden@9" to pin a snapshot
    task=my_task,
    scorers=["answer_correctness@2", exact_match],
    trial_count=3,
    max_concurrency=8,
)
print(result.url)

Platform scorers (strings) run server-side; @scorer functions run locally and are self-reported. A mixed list is split automatically.

Declaring an expected_output parameter is what makes a scorer reference-based. Cases without a reference skip that dimension and are counted rather than scored wrongly.

Raise Skip("reason") from a task to decline a case. An uncaught exception is retried three times with exponential backoff, then recorded as an error — the run continues either way, so one bad case never costs you the cohort.

Gate CI on it

result.wait_for_scoring()
result.assert_no_regression("exp_abc123", dimensions=["answer_correctness"])

assert_no_regression raises an AssertionError subclass, so an unhandled failure exits non-zero. Inconclusive cases fail by default: a comparison that could not reach a verdict is not evidence of safety. Pass allow_inconclusive=True once you have decided that risk is acceptable, and allow_scoring_errors=True to accept a run whose scoring finished with errors.

Scoring is asynchronous, so wait_for_scoring() comes first — the assertion refuses to guess from partial results rather than passing by accident.

Resuming

experiment.run(..., resume="exp_abc123")

Only a failed experiment can resume, and only against the same task fingerprint. If the code changed, the SDK refuses: a cohort half-answered by two versions supports no conclusion. Successes and skips are left untouched; missing and errored slots are rerun.

Examples

Run any of these examples to see the SDK in action. First, ensure environment variables are set:

# Traces with OpenAI
python examples/openai_example.py

# Logs with standard Python logging
python examples/logs_example.py

# Metrics (counters, histograms, up/down counters)
python examples/metrics_example.py

# LangChain Q&A with session tracking
python examples/session_demo.py

See the examples/ directory for more samples including LangChain RAG chains and user tracking patterns.

Contributing

We welcome contributions! Please feel free to open issues or submit pull requests on GitHub.

Support

• 📖 OpenObserve Documentation
• 🐛 Report Issues
• 💬 OpenObserve Community

License

MIT