Python Flask instrumentation
Send Flask traces and logs to Observability Unified with OpenTelemetry and the interaction header.
Use the Python OpenTelemetry SDK when instrumenting Flask services. Observability Unified accepts standard OTLP/HTTP, so Python services can send traces and logs to the same collector as the first-party TypeScript SDKs.
Install packages
python -m pip install \
opentelemetry-api \
opentelemetry-sdk \
opentelemetry-exporter-otlp-proto-http \
opentelemetry-instrumentation-flask \
opentelemetry-instrumentation-requests \
flaskConfigure tracing
import os
from flask import Flask, request
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.instrumentation.flask import FlaskInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor
from opentelemetry.sdk.resources import Resource
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
collector_url = os.environ.get("OBS_COLLECTOR_URL", "http://localhost:8790")
ingest_key = os.environ["OBS_INGEST_KEY"]
provider = TracerProvider(
resource=Resource.create({
"service.name": "flask-api",
"service.version": "1.0.0",
})
)
provider.add_span_processor(
BatchSpanProcessor(
OTLPSpanExporter(
endpoint=f"{collector_url}/v1/traces",
headers={"authorization": f"Bearer {ingest_key}"},
)
)
)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("flask-api")
app = Flask(__name__)
FlaskInstrumentor().instrument_app(app)
RequestsInstrumentor().instrument()Stamp browser interaction ids
When a request comes from a browser instrumented with @obsunified/analytics-sdk, the request carries x-obs-interaction. Copy that value onto the active span as obs.interaction.id.
@app.before_request
def stamp_interaction_id():
interaction_id = request.headers.get("x-obs-interaction")
if not interaction_id:
return
span = trace.get_current_span()
if span and span.is_recording():
span.set_attribute("obs.interaction.id", interaction_id)Create child spans
@app.get("/api/recommendations")
def recommendations():
with tracer.start_as_current_span("recommendations.query") as span:
span.set_attribute("db.system", "postgresql")
rows = load_recommendations()
return {"items": rows}Verify
Start your collector, set the environment, then call the Flask route from an instrumented browser or with a manual header:
export OBS_COLLECTOR_URL=http://localhost:8790
export OBS_INGEST_KEY=<your key> # the collector's INGEST_KEY, or a key from `obs-unified keys mint`
flask --app app run --port 5001
curl -H 'x-obs-interaction: 01JAQ8K3Z9X7M2N4P6Q8R0S1T3' http://localhost:5001/api/recommendationsNotes:
- Port 5001 is used because macOS reserves port 5000 for AirPlay Receiver on many systems. Any free port works (for example
--port 8000). OBS_INGEST_KEYmust be a real key accepted by your collector. The prebuilt GHCR image rejects the sampledev-ingest-key; use the value you passed asINGEST_KEY.x-obs-interactionvalues are 26-character uppercase Crockford-base32 ULIDs. The browser SDK mints valid ids, and the TypeScript server SDK drops malformed values. This Flask hook stores whatever string it receives, but the interaction-id contract shared by every SDK is a ULID, so use that format for manual tests rather than a free-form value likedemo_click_123.
The trace should appear in the Traces tab with obs.interaction.id, allowing the Connected Rail to pivot back to the originating browser interaction when one exists.