xr_ai_runtime

Typed in-process runtime for composable XR AI agents.

Exceptions

RuntimeClosedError

Raised when work is submitted to a runtime that is not running.

Classes

Agent	An object that owns state and exposes ordinary native tools.
MessageMetadata	Routing and trace context for one published event.
Topic	A stable publish/subscribe name paired with its payload model.
AgentRuntime	Provide typed pub/sub routing between registered agents.
RuntimeContext	Runtime operations available during a subscription delivery.

Functions

subscribe(...)

Register an agent method as a typed topic subscriber.

Package Contents

class xr_ai_runtime.Agent(

tools: collections.abc.Iterable[xr_ai_tools.Tool[Any, Any] | xr_ai_tools.AsyncTool[Any, Any]] = (),

)

An object that owns state and exposes ordinary native tools.

property tools: tuple[xr_ai_tools.Tool[Any, Any] | xr_ai_tools.AsyncTool[Any, Any], Ellipsis]

Return the existing native tools exposed by this agent.

class xr_ai_runtime.MessageMetadata

Routing and trace context for one published event.

message_id: str

Unique identifier for this publication.

correlation_id: str

Identifier shared by publications in one logical operation.

participant_id: str | None

Participant routing key, or None for a global event.

source: str

Runtime-local name of the publisher.

parent_message_id: str | None

Identifier of the publication that caused this event, when present.

timestamp_us: int

Publication time as Unix microseconds.

class xr_ai_runtime.Topic

A stable publish/subscribe name paired with its payload model.

name: str

Stable name shared by every publisher and subscriber.

message_type: type[MessageT]

Pydantic model used to validate each publication.

telemetry: Literal['full', 'none'] = 'full'

Whether Relay records publication and subscriber-delivery scopes.

validate(message: MessageT | dict[str, Any]) → MessageT

Validate a message before delivery.

xr_ai_runtime.subscribe(

topic: Topic[MessageT],

) → collections.abc.Callable[[collections.abc.Callable[Ellipsis, collections.abc.Awaitable[None]]], collections.abc.Callable[Ellipsis, collections.abc.Awaitable[None]]]

Register an agent method as a typed topic subscriber.

class xr_ai_runtime.AgentRuntime

Provide typed pub/sub routing between registered agents.

property running: bool

Whether the runtime currently accepts event work.

register(name: str, agent: AgentT) → AgentT

Register one agent before startup and return the same typed object.

async start() → None

Start accepting event publications.

async publish(

topic: xr_ai_runtime.events.Topic[MessageT],

message: MessageT | dict[str, Any],

*,

participant_id: str | None = None,

source: str = 'application',

) → None

Validate and deliver one event to every topic subscriber.

async stop() → None

Stop accepting events and cancel in-flight deliveries.

exception xr_ai_runtime.RuntimeClosedError

Raised when work is submitted to a runtime that is not running.

class xr_ai_runtime.RuntimeContext(

runtime: AgentRuntime,

agent_name: str,

metadata: xr_ai_runtime.events.MessageMetadata,

)

Runtime operations available during a subscription delivery.

property agent_name: str

Return this agent’s runtime-local name.

property metadata: xr_ai_runtime.events.MessageMetadata

Return metadata for the current subscription delivery.

async publish(

topic: xr_ai_runtime.events.Topic[MessageT],

message: MessageT | dict[str, Any],

*,

participant_id: str | None = None,

) → None

Publish an event, preserving delivery scope when one exists.