mobius-tracer-py

Request-context propagation for Mobius Python (FastAPI / Starlette) services.

On every incoming request it extracts identity/trace headers and the JWT payload, exposes them request-scoped, forwards them onto downstream calls, sets security + trace response headers, and enriches OpenTelemetry spans.

• PyPI: pip install mobius-tracer-py
• Import package: mobius_tracer
• Python: 3.10+

Install

pip install mobius-tracer-py

# optional features
pip install "mobius-tracer-py[otel]"        # set OTel span attributes
pip install "mobius-tracer-py[httpx]"       # TracedClient / httpx hook
pip install "mobius-tracer-py[resources]"   # CPU/mem span metrics (psutil)

Quick start

from fastapi import FastAPI
from mobius_tracer import setup_tracer, current

app = FastAPI()
setup_tracer(app)   # adds middleware + 401 handler for invalid tokens

@app.get("/me")
async def me():
    ctx = current()
    return {"tenant_id": ctx.tenant_id, "txn": ctx.req_transaction_id}

setup_tracer installs a middleware that, per request:

1. extracts x-request-id, x-b3-traceid, x-b3-spanid, Authorization, appId, platformId, and X-Transaction-Id (generated if absent);
2. parses the JWT payload into the request context (tenant/agent/user/email …);
3. derives action_log_user_id from the token;
4. sets security response headers + x-mb-trace-id + X-Transaction-Id;
5. sets OpenTelemetry span attributes (txn.id, tenant.id, user.id).

Options

setup_tracer(
    app,
    skip_paths=("actuator", "health", "metrics", "error", "prometheus",
                "swagger", "api-docs", "arazzo"),  # paths with no token check
    require_token=True,          # 401 on protected paths without a valid token
    set_security_headers=True,   # HSTS/CSP/X-Frame-Options/...
    propagate_to_otel=True,      # txn/tenant/user span attributes
    integrate_logging=True,      # mirror fields into mobius-logging-py context
    measure_resources=True,      # CPU/mem of each request onto the span
    instrument_httpx=False,      # patch httpx so all outbound calls propagate (opt-in)
)

Request context

from mobius_tracer import current, get, HeadersHolder

ctx = current()           # HeadersHolder for this request
ctx.tenant_id             # e.g. "tenant-123"
ctx.action_log_user_id    # derived user id
get("trace_id")           # field accessor

Fields: request_id, trace_id, span_id, tenant_id, tenant_user_id, consumer_id, email, authorization, name, requester_type, platform_id, parent_tenant_id, app_id, action_log_user_id, agent_id, agent_user_id, req_transaction_id, b_agent_id.

The context uses contextvars, so it is isolated per request and correct across async tasks and threads.

Propagating to downstream calls

Forward the current context (and a W3C traceparent) to outbound requests:

import httpx
from mobius_tracer import traced_headers, httpx_auth_hook, TracedClient

# 1) build headers yourself
httpx.get(url, headers=traced_headers())

# 2) an httpx event hook
client = httpx.AsyncClient(event_hooks={"request": [httpx_auth_hook()]})

# 3) a ready-made traced client
with TracedClient(base_url="https://api.internal") as c:
    c.get("/orders/1")

# 4) globally patch httpx so ALL outbound calls propagate (internal-only!)
from mobius_tracer import instrument_httpx
instrument_httpx()   # or setup_tracer(app, instrument_httpx=True)

Forwarded headers: x-request-id, x-b3-traceid, x-b3-spanid, tenantId, tenantUserId, consumerId, email, name, Authorization, platformId, parentTenantId, appId, X-Transaction-Id, plus traceparent.

JWT parsing

from mobius_tracer import parse_token, TokenError

body = parse_token(jwt_string)   # decodes payload (no signature verification)
body.tenant_id, body.requester_type, body.email

Raises TokenError (HTTP 401) for missing or malformed tokens. A non-empty requester_type claim is required.

Resource metrics

With measure_resources=True (default), every request is measured and the metrics are attached to its OpenTelemetry span automatically. For a specific function or block:

from mobius_tracer import measure_resources, measure

@measure_resources
async def heavy(): ...

with measure():
    do_work()

Span attributes: wall_time_ms, cpu_used_ms, cpu_used_pct, cpu_cores, mem_rss_mb, mem_rss_delta_mb, mem_system_used_pct. Best-effort; never raises. Uses psutil when installed.

Integrating with mobius-logging-py

setup_tracer(app, integrate_logging=True)

When mobius-logging-py is installed, the middleware mirrors tenant_id, trace_id, txn_id, agent_id, and correlation_id into its logging context, so every log line carries the same identity fields.

Development

python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"
pytest