Gateway Tracing Hook

Emit OpenTelemetry-compatible spans across the gateway pipeline via a zero-cost core seam

from praisonaiagents import Agent

agent = Agent(name="Support", instructions="Help the customer.")
# The gateway wraps each pipeline stage in a span your tracer can pick up
agent.start("Where is my order?")

Wrap each stage of the gateway pipeline in a span your own OpenTelemetry exporter can pick up — and carry W3C trace context (traceparent / tracestate) across process boundaries so the trace tree doesn’t break at the gateway.

The user sends a turn; the gateway opens a span around every stage — inbound, admission, agent run, each LLM and tool call, outbox, delivery — so latency and failures show up per stage in Jaeger, Tempo, Datadog, or Honeycomb.

When an exporter is attached, an inbound traceparent from a caller becomes the parent of agent.run, and the active context is propagated to downstream LLM / MCP / tool calls.

Quick Start

1. Simple Usage — the safe, zero-cost default

   Do nothing and the gateway holds a no-op hook. Every stage(...) returns a null context manager that ignores its arguments, so tracing adds negligible overhead until you attach a real exporter.

   from praisonaiagents.gateway import NULL_GATEWAY_TRACE_HOOK
   
   # The default a gateway uses until an exporter is supplied — one shared, stateless instance
   with NULL_GATEWAY_TRACE_HOOK.stage("agent.run", correlation_id="turn-42", session="s1") as span:
       assert span is None   # no-op: nothing is recorded, nothing is swallowed

2. With a real tracer — plug a custom hook

   Any object with a matching stage(name, *, correlation_id, **attrs) context-manager factory satisfies the protocol — no base class needed. This recorder captures each stage as it opens and closes:

   from contextlib import contextmanager
   from praisonaiagents.gateway import resolve_trace_hook
   
   class Recorder:
       def __init__(self):
           self.events = []
   
       def stage(self, name, *, correlation_id=None, **attrs):
           return self._span(name, correlation_id, attrs)
   
       @contextmanager
       def _span(self, name, correlation_id, attrs):
           self.events.append(("start", name, correlation_id))
           try:
               yield name
           finally:
               self.events.append(("end", name, correlation_id))
   
   tracer = resolve_trace_hook(Recorder())
   
   with tracer.stage("agent.run", correlation_id="c1", session="s1") as span:
       ...  # run the stage; the span opens here and closes on exit

   A production exporter follows the same shape but calls tracer.start_as_current_span(...) from the OpenTelemetry SDK inside stage. That exporter lives in the separate praisonai-plugins package — core ships only the seam and the no-op default.

3. Continue an inbound trace — parent agent.run under the caller

   An inbound request arrives carrying a traceparent; pull it out with extract_carrier and pass it as parent_carrier so agent.run continues the caller’s trace instead of starting a fresh detached one.

   from praisonaiagents import Agent
   from praisonaiagents.gateway import resolve_trace_hook
   
   tracer = resolve_trace_hook(None)  # swap for a real exporter in production
   
   # inbound.headers e.g. {"traceparent": "00-...-...-01", "tracestate": "..."}
   parent = tracer.extract_carrier(inbound.headers)
   
   agent = Agent(name="Support", instructions="Help the customer.")
   
   with tracer.stage("agent.run", correlation_id="turn-42",
                     parent_carrier=parent, session="s1"):
       reply = agent.start("Where is my order?")

   With the no-op default, extract_carrier returns None and stage ignores parent_carrier — the code is safe to write before you attach an exporter.

How It Works

The hook is a synchronous context-manager factory, so it wraps both sync and async stages with the same with ...: block.

With W3C trace-context propagation

When an exporter is attached, the gateway continues the caller’s trace on ingress and re-injects the active context on egress, so a single distributed trace flows inbound → agent.run → tool / mcp / llm across processes.

With the no-op default, extract_carrier returns None and inject_context leaves the carrier untouched, so the same code runs unchanged with no exporter attached — agent.run simply starts a fresh root span.

Configuration Options

Core exposes the seam through five top-level symbols on praisonaiagents.gateway:

from praisonaiagents.gateway import (
    GatewayTraceHook,
    NullGatewayTraceHook,
    NULL_GATEWAY_TRACE_HOOK,
    resolve_trace_hook,
    GATEWAY_TRACE_STAGES,
)

The stage contract is deliberately dependency-free — no OpenTelemetry import lives in core:

def stage(
    self,
    name: str,
    *,
    correlation_id: Optional[str] = None,
    parent_carrier: Optional[Mapping[str, str]] = None,   # NEW
    **attrs: Any,
) -> AbstractContextManager[Any]:
    ...

Propagation methods (new in #4103)

Two additive methods move W3C trace context across process boundaries — one at egress, one at ingress:

def inject_context(self, carrier: MutableMapping[str, str]) -> None:
    """Write the active span context into `carrier` as W3C headers."""

def extract_carrier(
    self, carrier: Mapping[str, str]
) -> Optional[Mapping[str, str]]:
    """Return a propagation carrier extracted from inbound `carrier`, or None."""

Canonical Stage Names

GATEWAY_TRACE_STAGES is a tuple of the seven canonical span names, so a tracer plugin and the wrapper agree on names without importing each other.

Common Patterns

Pattern 1: Correlation id as a span attribute

Pass the inbound turn’s existing correlation id so spans and logs share a single key — the same id you already join logs on.

from praisonaiagents.gateway import NULL_GATEWAY_TRACE_HOOK

trace = NULL_GATEWAY_TRACE_HOOK  # swap for a real exporter in production

with trace.stage("delivery", correlation_id="turn-42", channel="telegram"):
    ...  # deliver the reply; the correlation id lands on the span as an attribute

Pattern 2: Wrapping a custom stage

Fire the seam around an async stage with the same synchronous with block:

with self._trace.stage("agent.run", correlation_id=current_correlation_id(), session=sid):
    reply = await agent.astart(text)

Pattern 3: resolve_trace_hook in a constructor

Accept an optional tracer= and resolve it once, so no stage ever branches on None:

from praisonaiagents.gateway import resolve_trace_hook

class MyGateway:
    def __init__(self, tracer=None):
        self._trace = resolve_trace_hook(tracer)  # a callable hook either way

    async def handle(self, text, sid):
        with self._trace.stage("agent.run", correlation_id="c1", session=sid):
            ...

Pattern 4: Continue a caller’s trace on ingress

Extract the inbound carrier once and pass it as parent_carrier, so agent.run nests under the caller’s span:

# inbound.headers e.g. {"traceparent": "00-...", "tracestate": "..."}
parent = self._trace.extract_carrier(inbound.headers)

with self._trace.stage(
    "agent.run",
    correlation_id=current_correlation_id(),
    parent_carrier=parent,   # None with the no-op default; agent.run then starts a fresh root
    session=sid,
):
    reply = await agent.astart(text)

End to end the user sees one distributed trace — inbound → agent.run → tool / mcp / llm — where the agent turn hangs off the caller’s span instead of starting over.

Pattern 5: Inject the active context on egress

Seed the outbound headers with the provider’s own headers, then inject so the active context wins over any stale traceparent:

# LLM / MCP / HTTP-tool egress: propagate the active trace to downstream services.
# Seed with provider headers first, then inject so the active context wins over
# any stale traceparent / tracestate.
headers = dict(provider_headers)
self._trace.inject_context(headers)  # no-op unless an exporter is attached
await http.post(url, headers=headers, json=payload)

End to end the downstream service’s span nests under agent.run, so the same distributed trace continues across the process boundary instead of breaking at the egress call.

Best Practices

Choosing an Integration Path

Related