ant_ai.observer.protocol

ObservabilitySink

Bases: Protocol

Protocol for observability backends.

Implementations must be safe to call concurrently and must never raise — wrap any internal errors in a try/except.

Source code in src/ant_ai/observer/protocol.py

class ObservabilitySink(Protocol):
    """Protocol for observability backends.

    Implementations must be safe to call concurrently and must never raise —
    wrap any internal errors in a try/except.
    """

    async def event(self, name: str, **fields: object) -> None:
        """Emit a named lifecycle event with structured metadata.

        Event names follow dot-namespaced conventions, e.g.
        `workflow.start`, `node.error`, `step.end`.

        Args:
            name: Dot-namespaced event name.
            **fields: Arbitrary structured metadata to attach to the event.
        """
        ...

    async def exception(self, name: str, error: Exception, **fields: object) -> None:
        """Emit an error event carrying the exception instance.

        Tracing backends should call `span.record_exception`; log backends
        should log the exception alongside the structured fields.

        Args:
            name: Dot-namespaced event name.
            error: The exception to record.
            **fields: Arbitrary structured metadata to attach to the event.
        """
        ...

    def span(self, name: str, **attrs: object) -> AbstractAsyncContextManager:
        """Return an async context manager representing a unit of work.

        Span names use operation names, not lifecycle suffixes (e.g. `llm`,
        `tool`, `router`). Attributes carry per-operation metadata such as
        `model`, `tool_name`, `messages`, or `session_id`.

        The yielded object must expose `update(**fields)`: call sites (`llm_step`, `tool_step`) record the output and usage on it once the operation completes, and `obs.span`'s no-op fallback provides it too.
        Yielding a backend handle that lacks it -- or yielding nothing -- makes every such call site raise.

        Args:
            name: Operation name for the span.
            **attrs: Metadata attributes to attach to the span.

        Returns:
            An async context manager yielding an object with `update(**fields)`.
        """
        ...

    def propagation_headers(self) -> dict[str, str]:
        """Return headers encoding the current trace context for outbound requests.

        Used to propagate distributed trace context across process boundaries
        (e.g. into A2A HTTP calls). Returns an empty dict when tracing is
        inactive or not supported.
        """
        ...

    def attach_propagation_context(
        self, headers: dict[str, str]
    ) -> AbstractContextManager:
        """Return a context manager that restores a remote trace context.

        Extracts the trace context encoded in *headers* (e.g. from an inbound
        A2A request) and makes it the active context for the duration of the
        block, so child observations are nested under the remote parent span.

        Args:
            headers: Carrier dict previously produced by `propagation_headers`.
        """
        ...

event async

event(name: str, **fields: object) -> None

Emit a named lifecycle event with structured metadata.

Event names follow dot-namespaced conventions, e.g. workflow.start, node.error, step.end.

Parameters:

Name	Type	Description	Default
name	str	Dot-namespaced event name.	required
**fields	object	Arbitrary structured metadata to attach to the event.	{}

Source code in src/ant_ai/observer/protocol.py

async def event(self, name: str, **fields: object) -> None:
    """Emit a named lifecycle event with structured metadata.

    Event names follow dot-namespaced conventions, e.g.
    `workflow.start`, `node.error`, `step.end`.

    Args:
        name: Dot-namespaced event name.
        **fields: Arbitrary structured metadata to attach to the event.
    """
    ...

exception async

exception(
    name: str, error: Exception, **fields: object
) -> None

Emit an error event carrying the exception instance.

Tracing backends should call span.record_exception; log backends should log the exception alongside the structured fields.

Parameters:

Name	Type	Description	Default
name	str	Dot-namespaced event name.	required
error	Exception	The exception to record.	required
**fields	object	Arbitrary structured metadata to attach to the event.	{}

Source code in src/ant_ai/observer/protocol.py

async def exception(self, name: str, error: Exception, **fields: object) -> None:
    """Emit an error event carrying the exception instance.

    Tracing backends should call `span.record_exception`; log backends
    should log the exception alongside the structured fields.

    Args:
        name: Dot-namespaced event name.
        error: The exception to record.
        **fields: Arbitrary structured metadata to attach to the event.
    """
    ...

span

span(
    name: str, **attrs: object
) -> AbstractAsyncContextManager

Return an async context manager representing a unit of work.

Span names use operation names, not lifecycle suffixes (e.g. llm, tool, router). Attributes carry per-operation metadata such as model, tool_name, messages, or session_id.

The yielded object must expose update(**fields): call sites (llm_step, tool_step) record the output and usage on it once the operation completes, and obs.span's no-op fallback provides it too. Yielding a backend handle that lacks it -- or yielding nothing -- makes every such call site raise.

Parameters:

Name	Type	Description	Default
name	str	Operation name for the span.	required
**attrs	object	Metadata attributes to attach to the span.	{}

Returns:

Type	Description
AbstractAsyncContextManager	An async context manager yielding an object with update(**fields).

Source code in src/ant_ai/observer/protocol.py

def span(self, name: str, **attrs: object) -> AbstractAsyncContextManager:
    """Return an async context manager representing a unit of work.

    Span names use operation names, not lifecycle suffixes (e.g. `llm`,
    `tool`, `router`). Attributes carry per-operation metadata such as
    `model`, `tool_name`, `messages`, or `session_id`.

    The yielded object must expose `update(**fields)`: call sites (`llm_step`, `tool_step`) record the output and usage on it once the operation completes, and `obs.span`'s no-op fallback provides it too.
    Yielding a backend handle that lacks it -- or yielding nothing -- makes every such call site raise.

    Args:
        name: Operation name for the span.
        **attrs: Metadata attributes to attach to the span.

    Returns:
        An async context manager yielding an object with `update(**fields)`.
    """
    ...

propagation_headers

propagation_headers() -> dict[str, str]

Return headers encoding the current trace context for outbound requests.

Used to propagate distributed trace context across process boundaries (e.g. into A2A HTTP calls). Returns an empty dict when tracing is inactive or not supported.

Source code in src/ant_ai/observer/protocol.py

def propagation_headers(self) -> dict[str, str]:
    """Return headers encoding the current trace context for outbound requests.

    Used to propagate distributed trace context across process boundaries
    (e.g. into A2A HTTP calls). Returns an empty dict when tracing is
    inactive or not supported.
    """
    ...

attach_propagation_context

attach_propagation_context(
    headers: dict[str, str],
) -> AbstractContextManager

Return a context manager that restores a remote trace context.

Extracts the trace context encoded in headers (e.g. from an inbound A2A request) and makes it the active context for the duration of the block, so child observations are nested under the remote parent span.

Parameters:

Name	Type	Description	Default
headers	dict[str, str]	Carrier dict previously produced by propagation_headers.	required

Source code in src/ant_ai/observer/protocol.py

def attach_propagation_context(
    self, headers: dict[str, str]
) -> AbstractContextManager:
    """Return a context manager that restores a remote trace context.

    Extracts the trace context encoded in *headers* (e.g. from an inbound
    A2A request) and makes it the active context for the duration of the
    block, so child observations are nested under the remote parent span.

    Args:
        headers: Carrier dict previously produced by `propagation_headers`.
    """
    ...