Flask middleware — automatic HTTP request tracing

The zespan-flask package adds a Flask extension to your application that automatically creates a trace context for every incoming HTTP request via before_request/after_request/teardown_appcontext hooks. The @observe_llm and @observe_span decorators let you attach LLM call details and custom operation spans to that context without threading trace objects through your call stack.

Installation

pip install zespan-flask

Adding the extension

Import ZespanFlaskExtension and ZespanConfig, then instantiate the extension with your Flask app.

from flask import Flask
from zespan_flask import ZespanFlaskExtension, ZespanConfig

app = Flask(__name__)

config = ZespanConfig(api_key="zsp_your_api_key_here")
zespan_ext = ZespanFlaskExtension(app, config)

ZespanFlaskExtension is a Flask extension (the same pattern used by Flask-SQLAlchemy and similar libraries), not a WSGI-callable wrapper — pass it your Flask app instance directly, not app.wsgi_app. It supports the app-factory pattern too: construct it with just the config (zespan_ext = ZespanFlaskExtension(config=config)) and call zespan_ext.init_app(app) once the app is created. Once registered, every HTTP request to your app creates a trace context. The extension records the HTTP method, path, response status code, and total request duration, then flushes all span data to Zespan asynchronously.

`ZespanConfig` options

string

required

Your Zespan API key. Must start with zsp_. Find it in Settings → API Keys.

string

default:"https://api.zespan.com"

Override the Zespan API base URL. Use this only if you are self-hosting the ingest endpoint.

boolean

default:"True"

When False, the middleware passes all requests through without tracing. Useful for disabling tracing in test environments.

float

default:"1.0"

Fraction of requests to trace, from 0.0 to 1.0. Set to 0.1 to trace 10% of requests in high-traffic environments.

boolean

default:"False"

When True, the request body is captured and attached to the trace. Enable only after reviewing your data retention policy.

boolean

default:"False"

When True, the response body is captured and attached to the trace.

list[str]

default:"[\"password\", \"token\", \"api_key\"]"

Field names whose values are redacted before storage. Applied regardless of capture_request_body.

boolean

default:"False"

When True, logs flush errors to stdout. Enable during integration testing.

Tracing LLM calls with `@observe_llm`

Use the @observe_llm decorator on any function that makes an LLM call. The decorator captures model name, provider, duration, and token usage (extracted automatically if the response has a .usage attribute).

from flask import Flask, request, jsonify
from zespan_flask import ZespanFlaskExtension, ZespanConfig, observe_llm
import openai

app = Flask(__name__)
zespan_ext = ZespanFlaskExtension(
    app,
    ZespanConfig(api_key="zsp_your_api_key_here"),
)

client = openai.OpenAI()

@observe_llm(model="gpt-4o", provider="openai")
def generate_summary(text: str) -> str:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "Summarize the following text concisely."},
            {"role": "user", "content": text},
        ],
    )
    return response.choices[0].message.content

@app.post("/summarize")
def summarize_endpoint():
    body = request.get_json()
    summary = generate_summary(body["text"])
    return jsonify({"summary": summary})

@observe_llm parameters:

string

Model identifier to record on the span, e.g. "gpt-4o" or "claude-sonnet-4-6".

string

Provider name, e.g. "openai" or "anthropic".

Tracing custom operations with `@observe_span`

Use @observe_span to instrument any function as a named span within the current request’s trace context.

from zespan_flask import observe_span

@observe_span("document-retrieval", span_type="retriever")
def retrieve_documents(query: str) -> list:
    # Vector store search
    results = vector_store.search(query, top_k=5)
    return [doc.content for doc in results]

@observe_span("rag-pipeline", span_type="custom")
def run_rag(query: str) -> str:
    docs = retrieve_documents(query)
    summary = generate_summary("\n\n".join(docs))
    return summary

string

required

The span name. Appears as the operation label in the trace flame graph.

string

default:"custom"

A hint for the span type. Common values: "llm", "retriever", "custom".

Setting custom attributes

Use set_attribute to attach arbitrary key-value data to the current request’s trace context from anywhere in your handler.

from zespan_flask import set_attribute

@app.post("/chat")
def chat_endpoint():
    body = request.get_json()
    set_attribute("user.id", body.get("user_id", "anonymous"))
    set_attribute("feature", "chat")

    response = run_rag(body["message"])
    return jsonify({"response": response})

Getting the current trace ID

Use get_current_trace_id() to retrieve the trace ID for the active request. Useful for correlating Zespan data with your own logging stack.

from zespan_flask import get_current_trace_id
import logging

logger = logging.getLogger(__name__)

@app.post("/generate")
def generate_endpoint():
    trace_id = get_current_trace_id()
    logger.info("Handling request", extra={"zespan_id": trace_id})

    result = generate_summary(request.get_json()["text"])
    return jsonify({"result": result, "trace_id": trace_id})

Complete example

from flask import Flask, request, jsonify
from zespan_flask import (
    ZespanFlaskExtension,
    ZespanConfig,
    observe_llm,
    observe_span,
    set_attribute,
)
import openai

app = Flask(__name__)
zespan_ext = ZespanFlaskExtension(
    app,
    ZespanConfig(
        api_key="zsp_your_api_key_here",
        sample_rate=1.0,
    ),
)

openai_client = openai.OpenAI()

@observe_span("vector-search", span_type="retriever")
def search_docs(query: str) -> list:
    return ["Relevant document 1", "Relevant document 2"]

@observe_llm(model="gpt-4o", provider="openai")
def answer_question(question: str, context: str) -> str:
    response = openai_client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": f"Answer using this context:\n{context}"},
            {"role": "user", "content": question},
        ],
    )
    return response.choices[0].message.content

@app.post("/ask")
def ask_endpoint():
    body = request.get_json()
    set_attribute("user.id", body.get("user_id", "anonymous"))

    docs = search_docs(body["question"])
    answer = answer_question(body["question"], "\n".join(docs))
    return jsonify({"answer": answer})

if __name__ == "__main__":
    app.run(debug=True)

The extension flushes queued traces synchronously (a blocking HTTP call) once the queue reaches 10 items or the flush interval elapses, and again on teardown_appcontext. For high-traffic applications, run behind a WSGI server (Gunicorn, uWSGI) that handles request concurrency so an occasional flush doesn’t block response delivery for other in-flight requests.

For async Flask applications (using flask[async]), the @observe_llm and @observe_span decorators support both sync and async functions transparently.

Overview

TypeScript SDK

Python SDK

Advanced SDK Configuration

Integrations

LLM Providers

Agent Frameworks

RAG Frameworks

Vector Databases

Custom / Other

Guides

Flask middleware — automatic HTTP request tracing

Installation

Adding the extension

`ZespanConfig` options

Tracing LLM calls with `@observe_llm`

Tracing custom operations with `@observe_span`

Setting custom attributes

Getting the current trace ID

Complete example

​Installation

​Adding the extension

​ZespanConfig options

​Tracing LLM calls with @observe_llm

​Tracing custom operations with @observe_span

​Setting custom attributes

​Getting the current trace ID

​Complete example

Installation

Adding the extension

`ZespanConfig` options

Tracing LLM calls with `@observe_llm`

Tracing custom operations with `@observe_span`

Setting custom attributes

Getting the current trace ID

Complete example