flowlines 0.2.2

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, Google Generative AI (Gemini), 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, google-generativeai, 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:

import flowlines

flowlines.init(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(...)

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.init(
    api_key="your-flowlines-api-key",
    ingest_endpoint="https://your-custom-endpoint.example.com",
)

You can also override the API endpoint used for memory retrieval and session management:

flowlines.init(
    api_key="your-flowlines-api-key",
    ingest_endpoint="https://your-custom-ingest.example.com",
    api_endpoint="https://your-custom-api.example.com",
)

Both endpoints must use HTTPS, unless they target localhost / 127.0.0.1 / ::1 (useful for local development).

Memory retrieval

Retrieve user memory from the Flowlines backend with get_memory() (sync) or aget_memory() (async):

memory = flowlines.get_memory("user-42")
memory = flowlines.get_memory("user-42", session_id="sess-abc", agent_id="agent-1", view="summary")

memory = await flowlines.aget_memory("user-42")

Both return a JSON string of the memory object, or None if no memory was found (HTTP 404). On other errors, they raise FlowlinesMemoryError.

Session management

Signal that a session has ended with end_session() (sync) or aend_session() (async). This flushes pending spans before notifying the backend:

flowlines.end_session("user-42", session_id="sess-abc")

await flowlines.aend_session("user-42", session_id="sess-abc")

On errors, they raise FlowlinesSessionError.

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:

import flowlines

flowlines.init(
    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.init(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. flowlines.init() 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.

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