Framework Integrations

Built-in middleware and interceptors wrap each request or task in a story() automatically, so handlers only need to call stage().

FastAPI / Starlette

RuntimeNarrativeMiddleware wraps every HTTP request in async with story("METHOD /path"). Route handlers receive the story context automatically and call stage() directly — no story() block needed in handlers.

python

from fastapi import FastAPI
from runtime_narrative import RuntimeNarrativeMiddleware, stage
from runtime_narrative.renderer.json_renderer import JsonRenderer

app = FastAPI()

app.add_middleware(
    RuntimeNarrativeMiddleware,
    renderers=[JsonRenderer()],
    failure_diagnostics="rich",
    app_roots=["myapp"],
)

@app.post("/orders")
async def create_order(payload: dict):
    async with stage("Validate Input"):
        assert payload.get("item_id"), "item_id required"
    async with stage("Persist Order"):
        return {"order_id": 42}

Auto-renderer selection

When renderers is omitted, the middleware checks sys.stdout.isatty(): TTY → ConsoleRenderer (local dev), non-TTY → JsonRenderer (Docker/CI/production). See Environment Variables for the full RUNTIME_NARRATIVE_RICH_LOG_FILE / RUNTIME_NARRATIVE_RICH_LOG_CONSOLE reference — every auto-instrumentation entry point (HTTP middleware, Django, Celery, gRPC) shares one default_renderers() selector, so the same env vars apply everywhere on this page.

HTTP status folded into the story line

The middleware records the response status as the story's outcome (StoryCompleted.outcome, e.g. "200 OK"). ConsoleRenderer then emits one self-contained line per request:

text

[d7678e] ▶ Story ended: GET /api/call - SUCCESS (200 OK, 0.023s)

This replaces the server access log — run uvicorn with access_log=False to avoid the duplicate INFO: "GET /api/call" 200 OK line. The same applies to the Django middlewares below. To set an outcome manually, call runtime.set_outcome(http_outcome(status)) (both importable from runtime_narrative) on any story.

W3C traceparent propagation

The middleware reads traceparent/tracestate headers from each incoming request and attaches the extracted OTel context before creating the story span. Story spans become children of the caller's trace when combined with OtelRenderer.

python

app.add_middleware(
    RuntimeNarrativeMiddleware,
    renderers=[OtelRenderer(tracer_provider=my_provider)],
    propagate_trace_context=True,   # default; set False to disable
)

All diagnostic kwargs supported by story() are forwarded: diagnostics_config, runtime_environment, failure_diagnostics, allow_rich_in_production, app_roots, redact_extra. Requires the fastapi extra.

Disabling instrumentation per-request

Pass a skip_if callable to suppress story creation for specific requests — useful for health checks, metrics scrapers, or test clients:

python

app.add_middleware(
    RuntimeNarrativeMiddleware,
    renderers=[JsonRenderer()],
    skip_if=lambda req: req.url.path in ("/health", "/metrics", "/readyz"),
)

When skip_if(request) returns True, the request passes straight through with no story created and no events emitted.

Run: uv run python examples/middleware_skip_if.py

Django

Two middleware classes are provided: an async ASGI variant and a sync WSGI variant. Both follow the same story naming convention ("GET /path", "POST /path").

Async ASGI

python

# settings.py
MIDDLEWARE = [
    "runtime_narrative.middleware_django.RuntimeNarrativeDjangoMiddleware",
    "django.middleware.common.CommonMiddleware",
    # ...
]

The async middleware uses async with story(...), so async renderers are awaited and AlertRoutingRenderer works correctly.

Sync WSGI

python

MIDDLEWARE = [
    "runtime_narrative.middleware_django.RuntimeNarrativeDjangoSyncMiddleware",
    # ...
]

Passing options

Django does not support constructor kwargs in MIDDLEWARE. Subclass to inject options:

python

# myapp/middleware.py
from runtime_narrative.middleware_django import RuntimeNarrativeDjangoMiddleware
from runtime_narrative.renderer.json_renderer import RotatingJsonRenderer

class AppNarrativeMiddleware(RuntimeNarrativeDjangoMiddleware):
    def __init__(self, get_response):
        super().__init__(
            get_response,
            renderers=[RotatingJsonRenderer("narrative.log")],
            failure_diagnostics="rich",
        )

Both classes auto-select ConsoleRenderer on TTY, JsonRenderer otherwise, when renderers is omitted — the same shared default_renderers() env vars described above apply here too. Requires the django extra.

Celery

Using NarrativeTask as a base class:

python

from celery import Celery
from runtime_narrative.celery import NarrativeTask
from runtime_narrative import stage

app = Celery("myapp", broker="redis://localhost:6379/0")

@app.task(base=NarrativeTask)
def process_invoice(invoice_id: int) -> dict:
    with stage("Fetch Invoice"):
        invoice = {"id": invoice_id, "amount": 100}
    with stage("Calculate Tax"):
        invoice["tax"] = invoice["amount"] * 0.1
    return invoice

The story name is "myapp.tasks.process_invoice [task_id=<celery-task-id>]". Configure renderers via class-level attributes:

python

from runtime_narrative.celery import NarrativeTask
from runtime_narrative.renderer.json_renderer import RotatingJsonRenderer

class MyTask(NarrativeTask):
    narrative_renderers = [RotatingJsonRenderer("celery_narrative.log")]
    narrative_failure_diagnostics = "rich"
    narrative_app_roots = ["myapp"]

@app.task(base=MyTask)
def process_invoice(invoice_id: int) -> dict:
    ...

Using connect_narrative to configure all tasks globally:

python

from runtime_narrative.celery import NarrativeTask, connect_narrative
from runtime_narrative.renderer.json_renderer import RotatingJsonRenderer

connect_narrative(
    app,
    renderers=[RotatingJsonRenderer("celery_narrative.log")],
    failure_diagnostics="rich",
    app_roots=["myapp"],
)

# All tasks using base=NarrativeTask now use these settings
@app.task(base=NarrativeTask)
def send_email(user_id: int) -> None: ...

@app.task(base=NarrativeTask)
def generate_report(report_id: int) -> None: ...

NarrativeTask uses with story(...) (sync). Async renderers are not awaited. Requires the celery extra.

gRPC

Sync interceptor:

python

import grpc
from concurrent import futures
from runtime_narrative.grpc_interceptor import RuntimeNarrativeInterceptor
from runtime_narrative.renderer.json_renderer import JsonRenderer

interceptor = RuntimeNarrativeInterceptor(
    renderers=[JsonRenderer()],
    failure_diagnostics="rich",
)

server = grpc.server(
    futures.ThreadPoolExecutor(max_workers=10),
    interceptors=[interceptor],
)

Inside a servicer, call stage() without wrapping in your own story():

python

from runtime_narrative import stage

class OrderServicer(orders_pb2_grpc.OrderServiceServicer):
    def CreateOrder(self, request, context):
        with stage("Validate"):
            assert request.item_id, "item_id required"
        with stage("Persist"):
            return orders_pb2.OrderResponse(order_id=42)

Async interceptor:

python

import grpc.aio
from runtime_narrative.grpc_interceptor import RuntimeNarrativeAsyncInterceptor

interceptor = RuntimeNarrativeAsyncInterceptor(
    renderers=[JsonRenderer()],
)

server = grpc.aio.server(interceptors=[interceptor])

The async interceptor uses async with story(...), so async renderers are awaited. Requires the grpc extra.

Next step

See Async Task Groups for running concurrent work inside a single story, whether or not it's driven by one of these integrations.