Custom Renderers

Any object with a handle method is a valid renderer. There is no protocol to import, no base class to extend.

Sync renderer

A sync renderer's handle method is called directly by StoryRuntime.emit(). It works with both sync and async stories.

python

from runtime_narrative import story, stage

class MetricsRenderer:
    def __init__(self):
        self.stage_durations: dict[str, float] = {}

    def handle(self, event: object) -> None:
        name = type(event).__name__
        if name == "StageCompleted":
            self.stage_durations[event.stage_name] = event.duration_seconds
        elif name == "FailureOccurred":
            print(f"[metrics] failure: {event.error_type} in {event.stage_name!r}")

metrics = MetricsRenderer()

with story("ETL", renderers=[metrics]):
    with stage("Extract"):
        pass
    with stage("Load"):
        pass

print(metrics.stage_durations)

Async renderer

An async renderer's handle coroutine is awaited by StoryRuntime.emit_async(). It is only awaited when using async with story(...). In a sync with story(...) context the async handle is never called.

python

import asyncio
import aiohttp
from runtime_narrative import story, stage

class WebhookRenderer:
    def __init__(self, url: str):
        self._url = url

    async def handle(self, event: object) -> None:
        if type(event).__name__ != "StageCompleted":
            return
        async with aiohttp.ClientSession() as session:
            await session.post(self._url, json={
                "stage": event.stage_name,
                "duration": event.duration_seconds,
            })

async def main():
    renderer = WebhookRenderer("https://collector.internal/stages")
    async with story("Pipeline", renderers=[renderer]):
        async with stage("Step 1"):
            await asyncio.sleep(0.01)

asyncio.run(main())

Inspecting event types

Use type(event).__name__ to branch on event types without importing from runtime_narrative.events. This avoids circular imports in third-party renderer packages.

python

def handle(self, event: object) -> None:
    name = type(event).__name__
    if name == "StoryStarted":
        self._start_timer(event.story_id)
    elif name == "StoryCompleted":
        elapsed = self._stop_timer(event.story_id)
        print(f"{event.story_name!r} took {elapsed:.2f}s, success={event.success}")
    elif name == "FailureOccurred":
        self._increment_counter(event.error_type)
    elif name == "LLMAnalysisReady":
        print(f"Analysis: {event.llm_analysis}")

If you prefer explicit imports:

python

from runtime_narrative.events import StoryStarted, StageCompleted, FailureOccurred

def handle(self, event: object) -> None:
    if isinstance(event, FailureOccurred):
        ...

Pass any object implementing handle as an entry in renderers=[...] to story() — see Renderers for the full set of built-in renderers this composes with.

Next step

Continue to Framework Integrations to see how renderers are wired into FastAPI, Django, Celery, and gRPC via middleware.