flowlines 0.2.4

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 include OpenAI, Anthropic, Google Generative AI (Gemini), AWS Bedrock, Cohere, Vertex AI, Together AI, Groq, Mistral AI, Ollama, and more. Also instruments frameworks and tools like LangChain, LlamaIndex, CrewAI, MCP, Pinecone, ChromaDB, Qdrant, and others.

Requirements

• Python 3.10+

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, groq, mistralai, ollama, replicate, transformers, sagemaker, watsonx, writer, alephalpha, voyageai, openai-agents, pinecone, chromadb, qdrant, lancedb, marqo, milvus, weaviate, langchain, llamaindex, crewai, agno, haystack, 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

Step 1: Initialize the SDK

Call flowlines.init() before creating any LLM client:

import flowlines
from openai import OpenAI

flowlines.init(api_key="your-flowlines-api-key")
client = OpenAI()

The SDK automatically detects which LLM libraries are installed, instruments them, and exports LLM-related spans to the Flowlines backend via OTLP/HTTP.

Step 2: Set user and session context

Wrap your LLM calls in flowlines.context() to tag spans with a user ID and session ID:

with flowlines.context(user_id="user-42", session_id="sess-abc"):
    response = client.chat.completions.create(model="gpt-4", messages=messages)

Step 3: Retrieve and inject memory

Retrieve what Flowlines remembers about a user from previous conversations:

memory = flowlines.get_memory("user-42")

get_memory() returns a JSON string, or None if no memory exists. Inject it into your prompt so the LLM can personalize its responses:

messages = [{"role": "system", "content": "You are a helpful assistant."}]
if memory:
    messages.append({
        "role": "system",
        "content": f"Here is what you know about this user from previous conversations:\n{memory}",
    })
messages.append({"role": "user", "content": "Hello!"})

Step 4: Make LLM calls

No changes needed — calls are auto-instrumented:

with flowlines.context(user_id="user-42", session_id="sess-abc"):
    response = client.chat.completions.create(model="gpt-4", messages=messages)

Step 5: End the session

When a conversation session is over, signal it to the backend. This flushes pending spans and notifies Flowlines:

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

Full example

import flowlines
from openai import OpenAI

flowlines.init(api_key="your-flowlines-api-key")
client = OpenAI()

user_id = "user-42"
session_id = "sess-abc"

# Retrieve memory for this user
memory = flowlines.get_memory(user_id)

messages = [{"role": "system", "content": "You are a helpful assistant."}]
if memory:
    messages.append({
        "role": "system",
        "content": f"Here is what you know about this user from previous conversations:\n{memory}",
    })
messages.append({"role": "user", "content": "Hello!"})

with flowlines.context(user_id=user_id, session_id=session_id):
    response = client.chat.completions.create(model="gpt-4", messages=messages)

flowlines.end_session(user_id=user_id, session_id=session_id)

Reference

Context 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.

Memory retrieval

Retrieve user memory 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 exists or if an error occurs. Errors are logged but never raised, so calling code is not disrupted.

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")

Errors are logged but never raised, so calling code is not disrupted.

Custom endpoints

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).

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

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

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