flowlines 0.1.0

Automatic instrumentation for LLM provider APIs — capture requests, responses, timing, and errors with minimal code changes.

Flowlines SDK for Python

Observability for LLM-powered applications. The Flowlines SDK instruments LLM provider APIs using OpenTelemetry — it captures requests, responses, timing, and errors, and exports them to the Flowlines backend via OTLP/HTTP.

Supported providers: OpenAI, Anthropic, AWS Bedrock, Cohere, Vertex AI, Together AI. Also instruments LangChain, LlamaIndex, MCP, Pinecone, ChromaDB, and Qdrant.

Requirements

• Python 3.11+

Installation

pip install flowlines

Then install instrumentation extras for the providers you use:

# Single provider
pip install flowlines[openai]

# Multiple providers
pip install flowlines[openai,anthropic]

# All supported providers
pip install flowlines[all]

Available extras: openai, anthropic, bedrock, cohere, vertexai, together, pinecone, chromadb, qdrant, langchain, llamaindex, mcp.

AI coding agent integration

If you use an AI coding agent, you can install the Flowlines skill so your agent knows how to integrate the SDK into your project:

npx skills add flowlines-ai/skills

Then, just ask your agent to integrate Flowlines into your project.

Quick start

If you don't have an existing OpenTelemetry setup, this is all you need:

from flowlines import Flowlines

flowlines = Flowlines(api_key="your-flowlines-api-key")

That's it. Every LLM call made through an installed provider is now automatically captured and exported to Flowlines. The SDK:

1. Creates an OpenTelemetry TracerProvider
2. Detects which LLM libraries are installed and instruments them
3. Filters spans to only export LLM-related telemetry
4. Sends data to the Flowlines backend via OTLP/HTTP

User, session, and agent tracking

Tag LLM calls with a user ID, session ID, and/or agent ID using the context() context manager:

with flowlines.context(user_id="user-42", session_id="sess-abc"):
    client.chat.completions.create(...)  # this span gets user_id and session_id
    client.chat.completions.create(...)  # same

You can also attach an agent_id to identify which agent produced the spans:

with flowlines.context(user_id="user-42", session_id="sess-abc", agent_id="agent-1"):
    client.chat.completions.create(...)

The session_id and agent_id are optional — you can track just the user:

with flowlines.context(user_id="user-42"):
    client.chat.completions.create(...)

For cases where a context manager doesn't fit (e.g. across request boundaries), use the imperative API. set_context() returns a token — pass it to clear_context() to restore the previous state:

token = Flowlines.set_context(user_id="user-42", session_id="sess-abc", agent_id="agent-1")
try:
    client.chat.completions.create(...)
finally:
    Flowlines.clear_context(token)

Context tracking is thread-safe and async-safe.

Custom endpoint

By default, data is sent to https://ingest.flowlines.ai. You can override this:

flowlines = Flowlines(
    api_key="your-flowlines-api-key",
    endpoint="https://your-custom-endpoint.example.com",
)

The endpoint must use HTTPS, unless it targets localhost / 127.0.0.1 / ::1 (useful for local development).

Usage with an existing OpenTelemetry setup

If your application already has its own TracerProvider, pass has_external_otel=True to prevent the SDK from creating a second one:

from flowlines import Flowlines

flowlines = Flowlines(
    api_key="your-flowlines-api-key",
    has_external_otel=True,
)

In this mode, the SDK does not create a TracerProvider or register instrumentors. You are responsible for wiring things up yourself:

from opentelemetry.sdk.trace import TracerProvider

provider = TracerProvider()

# 1. Add the Flowlines span processor to your provider
processor = flowlines.create_span_processor()
provider.add_span_processor(processor)

# 2. Instrument providers using the Flowlines instrumentor registry
for instrumentor in flowlines.get_instrumentors():
    instrumentor.instrument(tracer_provider=provider)

• create_span_processor() returns a span processor that filters and exports LLM spans to Flowlines. Call it exactly once.
• get_instrumentors() returns instrumentor instances for every supported provider library that is currently installed. You can also skip this and register instrumentors yourself.

Troubleshooting

Verbose mode

Pass verbose=True to enable debug logging. This prints detailed information about initialization, instrumentor discovery, span filtering, and export results to stderr:

flowlines = Flowlines(api_key="your-flowlines-api-key", verbose=True)

Example output:

[flowlines] Initializing Flowlines SDK (endpoint=https://ingest.flowlines.ai)
[flowlines] Mode A: creating TracerProvider and registering instrumentors
[flowlines] Instrumentor loaded: OpenAIInstrumentor
[flowlines] Instrumentor skipped: anthropic (library not installed)
[flowlines] Total instrumentors loaded: 1
[flowlines] Flowlines SDK initialized successfully
[flowlines] Export: 2/5 span(s) are LLM-related — sending to backend
[flowlines] Export: succeeded

No spans appearing in Flowlines

• Enable verbose mode. Pass verbose=True to see exactly what the SDK is doing — which instrumentors are loaded, how many spans are captured, and whether exports succeed.
• Missing instrumentation extras. The SDK only instruments providers whose instrumentation package is installed. For example, if you use OpenAI, make sure you installed flowlines[openai]. Check your installed packages with pip list | grep opentelemetry-instrumentation.
• Flowlines initialized too late. The Flowlines() constructor must run before any LLM calls. If the provider client is created before instrumentation is set up, those calls won't be captured.
• Wrong API key. Verify that the api_key you pass is valid. The SDK will export spans, but the backend will reject them silently if the key is invalid.

RuntimeError: Flowlines is a singleton

Flowlines enforces a single instance. You're calling Flowlines(...) more than once. Store the instance and reuse it, or initialize it once at application startup.

ValueError: Endpoint must use HTTPS

The SDK requires HTTPS for all endpoints except loopback addresses (localhost, 127.0.0.1, ::1). If you're testing locally, use http://localhost:<port>.

Spans are missing user_id / session_id

Make sure the LLM call happens inside the flowlines.context() block or between set_context() and clear_context(). If you're using threads or async tasks, note that context does not propagate automatically to child threads — set it in each task.

Duplicate spans or conflicting TracerProvider

If you already have an OpenTelemetry setup, you must pass has_external_otel=True. Otherwise the SDK creates its own TracerProvider, which conflicts with yours. See Usage with an existing OpenTelemetry setup.

Examples

See the examples/ directory for working sample applications:

• OpenAI conversational agent — Interactive agent with tool calling, demonstrating Mode A auto-instrumentation and context propagation.

License

MIT