> ## Documentation Index
> Fetch the complete documentation index at: https://docs.zespan.com/llms.txt
> Use this file to discover all available pages before exploring further.

# Flask middleware — automatic HTTP request tracing

> Add ZespanFlaskExtension to your Flask app to automatically trace every HTTP request and instrument LLM calls with @observe_llm and @observe_span decorators.

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

```bash theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
pip install zespan-flask
```

## Adding the extension

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

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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

<ParamField body="api_key" type="string" required>
  Your Zespan API key. Must start with `zsp_`. Find it in **Settings → API Keys**.
</ParamField>

<ParamField body="base_url" type="string" default="https://api.zespan.com">
  Override the Zespan API base URL. Use this only if you are self-hosting the ingest endpoint.
</ParamField>

<ParamField body="enabled" type="boolean" default="True">
  When `False`, the middleware passes all requests through without tracing. Useful for disabling tracing in test environments.
</ParamField>

<ParamField body="sample_rate" type="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.
</ParamField>

<ParamField body="capture_request_body" type="boolean" default="False">
  When `True`, the request body is captured and attached to the trace. Enable only after reviewing your data retention policy.
</ParamField>

<ParamField body="capture_response_body" type="boolean" default="False">
  When `True`, the response body is captured and attached to the trace.
</ParamField>

<ParamField body="redact_fields" type="list[str]" default="[&#x22;password&#x22;, &#x22;token&#x22;, &#x22;api_key&#x22;]">
  Field names whose values are redacted before storage. Applied regardless of `capture_request_body`.
</ParamField>

<ParamField body="debug" type="boolean" default="False">
  When `True`, logs flush errors to stdout. Enable during integration testing.
</ParamField>

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

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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:

<ParamField body="model" type="string">
  Model identifier to record on the span, e.g. `"gpt-4o"` or `"claude-sonnet-4-6"`.
</ParamField>

<ParamField body="provider" type="string">
  Provider name, e.g. `"openai"` or `"anthropic"`.
</ParamField>

## Tracing custom operations with `@observe_span`

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

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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
```

<ParamField body="name" type="string" required>
  The span name. Appears as the operation label in the trace flame graph.
</ParamField>

<ParamField body="span_type" type="string" default="custom">
  A hint for the span type. Common values: `"llm"`, `"retriever"`, `"custom"`.
</ParamField>

## Setting custom attributes

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

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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.

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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

```python theme={"theme":{"light":"github-light","dark":"one-dark-pro"}}
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)
```

<Warning>
  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.
</Warning>

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