xr-ai-hub#
xr-ai-hub-client is the minimal agent-side DeviceIOHub boundary. It depends
only on pyzmq and msgpack and contains no LiveKit, web-server, model, tool, or
voice-pipeline code. Refer to the generated API Reference for public types
and signatures.
Processor endpoints#
from xr_ai_hub import DataMessage, ProcessorEndpoint
endpoint = ProcessorEndpoint(
sub_addr="ipc:///tmp/xr_hub_pub",
push_addr="ipc:///tmp/xr_hub_in",
)
async def on_data(message: DataMessage) -> None:
print(message.participant_id, message.topic)
unsubscribe = endpoint.on_data(on_data)
await endpoint.run()
unsubscribe()
Participants are the subscription unit. With auto_subscribe=True, an endpoint
subscribes on join and unsubscribes on leave. Set its Subscribe filter to drop
unneeded data, audio, or frame signals at the ZMQ socket. Explicit-assignment
workers disable automatic subscription and call subscribe(participant_id).
run() requests the current roster so a worker started mid-session receives
ordinary joined events for participants already present. Join callbacks must
therefore be idempotent. Because ZMQ subscription changes are asynchronous,
workers that own readiness wait for wait_for_subscriptions() before announcing
availability.
Subscribe.REALTIME combines data, audio, and video. Subscribe.DEFAULT is
the same real-time-only filter. Subscribe.ALL remains an equivalent but
deprecated compatibility alias; completed files stay explicitly opt-in.
Completed files#
File transfer is opt-in and uses a bounded IPC lane separate from real-time
messages. Pass the file publisher address and include Subscribe.FILE, then
register an async callback:
from xr_ai_hub import FileMessage, ProcessorEndpoint, Subscribe
endpoint = ProcessorEndpoint(
sub_addr="ipc:///tmp/xr_hub_pub",
push_addr="ipc:///tmp/xr_hub_in",
file_sub_addr="ipc:///tmp/xr_hub_file_pub",
filter=Subscribe.REALTIME | Subscribe.FILE,
)
async def on_file(message: FileMessage) -> None:
print(message.participant_id, message.topic, message.name, len(message.data))
unsubscribe = endpoint.on_file(on_file)
The callback runs only after DeviceIOHub has received and validated the complete
LiveKit byte stream. File callbacks are awaited serially on their own receive
loop. Audio, lifecycle, and packet-data traffic continue on the real-time loop.
attributes contains only application metadata; StreamKit’s reserved routing
attributes are removed.
Frames and return routing#
Frame callbacks carry metadata without copying pixels. request_frame() obtains
the selected FrameData only on demand and coalesces concurrent requests for the
same participant and track. LiveFrameSource adds a fresh-frame cache and
participant cleanup. Conversion to opaque image references belongs in
xr_ai_tools.current_frame; VLM inference remains a separate vision tool.
Return data, audio, and flush messages always name their participant. Status is
agent state: set_status() and mark_ready() record it on the endpoint, while
DeviceIOHub aggregates responsible agents and publishes client readiness.
republish_statuses() restores one-shot state after reconnects.
Readiness participation is opt-in with announces_readiness=True. Passive
endpoints do not contribute status. An endpoint answers only for subscribed
participants, and a live subscription must be confirmed before the hub exposes
it as ready. agent_id comes from the constructor, XR_AI_AGENT_ID, or a
process-local generated value.