RoomKit

RoomKit

RoomKit(store=None, identity_resolver=None, identity_channel_types=None, inbound_router=None, lock_manager=None, realtime=None, max_chain_depth=5, agent_response_policy=AGENT_CHAIN, identity_timeout=10.0, process_timeout=30.0, stt=None, tts=None, voice=None, task_runner=None, delivery_strategy=None, delivery_backend=None, status_bus=None, telemetry=None, inbound_rate_limit=None, orchestration=None, persistence_policy=None, delivery_gap_timeout=30.0, delivery_claim_lock_manager=None)

Central orchestrator tying rooms, channels, hooks, and storage.

Initialise the RoomKit orchestrator.

Parameters:

Name	Type	Description	Default
store	ConversationStore | None	Persistent storage backend. Defaults to InMemoryStore.	None
identity_resolver	IdentityResolver | None	Optional resolver for identifying inbound senders. Without one, an event's participant_id is whatever the channel stamped on it (RFC §11.6) — so it is only as trustworthy as the transport that supplied it, and the edit/delete author check (RFC §10.3) is decided from it. Derive the sender id from an authenticated session rather than a client-supplied field, or install a resolver. See the Security Hardening guide.	None
identity_channel_types	set[ChannelType] | None	Restrict identity resolution to specific channel types. If None (default), resolution runs for all channels. Set to e.g. {ChannelType.SMS} to only resolve identity for SMS. Applies wherever resolution runs, the conference arrival path included — see :meth:identity_enabled_for.	None
inbound_router	InboundRoomRouter | None	Strategy for routing inbound messages to rooms. Defaults to DefaultInboundRoomRouter.	None
lock_manager	RoomLockManager | None	Per-room locking backend. Defaults to InMemoryLockManager. For multi-process deployments, supply a distributed implementation (e.g. Redis-backed).	None
realtime	RealtimeBackend | None	Realtime backend for ephemeral events (typing, presence). Defaults to InMemoryRealtime. For multi-process deployments, supply a distributed implementation (e.g. Redis pub/sub).	None
max_chain_depth	int	Maximum reentry chain depth to prevent infinite loops.	5
identity_timeout	float	Timeout in seconds for identity resolution calls.	10.0
process_timeout	float	Timeout in seconds for the locked processing phase.	30.0
stt	STTProvider | None	Optional speech-to-text provider for transcription.	None
tts	TTSProvider | None	Optional text-to-speech provider for synthesis.	None
voice	VoiceBackend | None	Optional voice backend for real-time audio transport.	None
task_runner	TaskRunner | None	Pluggable backend for delegated background tasks. Defaults to InMemoryTaskRunner.	None
delivery_strategy	DeliveryStrategy | str | None	Controls proactive delivery of background task results. When set, strategy.deliver() is called after system prompt injection and the ON_TASK_COMPLETED hook. Can be overridden per-task via delegate().	None
delivery_backend	DeliveryBackend | None	Persistent delivery backend. When set, kit.deliver() enqueues items instead of executing in-process, and a worker loop dequeues and executes them. Defaults to None (in-process delivery).	None
status_bus	StatusBus | None	Shared status bus for multi-agent coordination. Defaults to a StatusBus with InMemoryStatusBackend. Access via kit.status_bus.	None
telemetry	TelemetryConfig | TelemetryProvider | None	Optional telemetry provider or config for span/metric collection. Accepts a TelemetryProvider instance or a TelemetryConfig. Defaults to NoopTelemetryProvider.	None
inbound_rate_limit	RateLimit | None	Optional rate limit applied to all inbound messages before any processing. Messages exceeding the limit are dropped with reason="rate_limited". Keyed per channel_id.	None
orchestration	Orchestration | None	Default orchestration strategy applied to rooms created via create_room() unless overridden per-room.	None
persistence_policy	PersistencePolicy | None	Controls which event types are persisted. When None (default), all events are persisted. Use PersistencePolicy(exclude_types={...}) to skip specific types or PersistencePolicy(persist_types={...}) to whitelist.	None
delivery_gap_timeout	float	Seconds a delivery lane waits on a cursor hole owned by an absent worker (one that committed events and crashed before delivering them) before skipping over it with a delivery_skipped framework event (RFC §10.2). The skip bounds the loss to the crash window; it never covers an index this process holds a plan for.	30.0
delivery_claim_lock_manager	RoomLockManager | None	Lock manager used for the per-room delivery claims. Defaults to lock_manager. Supply a separate PostgresAdvisoryLockManager (own pool) in multi-process deployments: a claim is held for the length of a delivery set — provider round trips included — and must not starve the room-lock pool that commits depend on.	None

store property

store

The backing conversation store.

hook_engine property

hook_engine

The hook engine used for sync/async hook pipelines.

realtime property

realtime

The realtime backend for ephemeral events.

get_timeline async

get_timeline(room_id, offset=0, limit=50, visibility_filter=None, *, after_index=None, before_index=None, newest_first=False, organization_id=None)

Query the event timeline for a room.

Supports offset-based (offset/limit) and cursor-based (after_index/before_index) pagination. When a cursor parameter is set, offset is ignored.

The offset-based default is the oldest limit events — page 1 of a log reads from the beginning. Pass newest_first=True for the most recent limit instead (still ascending), which is the shape a reconnect snapshot wants: what was just said, not how the room opened.

Parameters:

Name	Type	Description	Default
room_id	str	Room to query.	required
offset	int	Number of events to skip (offset-based mode).	0
limit	int	Maximum number of events to return.	50
visibility_filter	str | None	Optional visibility value to filter by.	None
after_index	int | None	Return events with index > after_index.	None
before_index	int | None	Return events with index < before_index.	None
newest_first	bool	In offset-based mode, return the most recent limit events instead of the oldest. Ignored when a cursor is supplied.	False

list_tasks async

list_tasks(room_id, status=None, *, organization_id=None)

List tasks for a room, optionally filtered by status.

list_observations async

list_observations(room_id, *, organization_id=None)

List observations for a room.

send_event async

send_event(room_id, channel_id, content, event_type=MESSAGE, chain_depth=0, participant_id=None, metadata=None, visibility=ALL, provider=None, response_visibility=None, created_at=None, parent_event_id=None, idempotency_key=None, addressed_to=None, organization_id=None)

Send an event directly into a room from a channel.

Parameters:

Name	Type	Description	Default
room_id	str	Target room ID	required
channel_id	str	Source channel ID	required
content	Any	Event content (TextContent, RichContent, etc.)	required
event_type	EventType	Type of event (default MESSAGE)	MESSAGE
chain_depth	int	Depth in response chain (for loop prevention)	0
participant_id	str | None	Optional participant/sender ID for the event source	None
metadata	dict[str, Any] | None	Optional event metadata	None
visibility	str	Event visibility ("all" or "internal")	ALL
provider	str | None	Optional provider/backend name for event attribution	None
response_visibility	str | None	Controls where the AI's response is delivered. Uses the same vocabulary as visibility. None means no restriction.	None
addressed_to	list[str] | None	The intelligence channels this event asks to act (RFC §19.3). An address is set by a sender, and direct injection has a sender like any other entry point: a caller storing a message whose answer it triggers separately passes [] so the stored message solicits nobody, and one talking to a specific agent in a room that holds several names it. None leaves the event unaddressed — the router decides, or every eligible intelligence channel acts.	None
parent_event_id	str | None	In-app thread parent. The locked pipeline normalises it to the thread root (flat two-level model); see :meth:_resolve_thread_root.	None
idempotency_key	str | None	Stable de-duplication key. When set, the locked pipeline's idempotency check (backed by the unique events(room_id, idempotency_key) index) skips a re-send that carries the same key — so a caller that may replay a publish (an outbox dispatcher redelivering after a crash) gets at-most-once persistence for a given key. None keeps the prior behaviour (no de-duplication), matching inbound events that carry no key.	None

connect_websocket async

connect_websocket(channel_id, connection_id, send_fn, *, room_id, stream_send_fn=None)

Register a WebSocket connection for a room and emit framework event.

room_id says which conversation this socket belongs to. It is required because one channel instance can serve several rooms, and a connection the channel cannot place would otherwise have to receive every room's events. Add more rooms to the same socket with :meth:subscribe_websocket.

disconnect_websocket async

disconnect_websocket(channel_id, connection_id)

Unregister a WebSocket connection and emit framework event.

mark_read async

mark_read(room_id, channel_id, event_id, *, organization_id=None)

Mark an event as read for a channel.

mark_all_read async

mark_all_read(room_id, channel_id, *, organization_id=None)

Mark all events as read for a channel.

Exceptions

RoomNotFoundError

Bases: RoomKitError

Room does not exist.

ChannelNotFoundError

Bases: RoomKitError

Channel binding not found in room.

ChannelNotRegisteredError

Bases: RoomKitError

Channel type not registered.

Infrastructure

RoomLockManager

Bases: ABC

Abstract base for per-room locking.

Implement this to plug in any locking backend (Redis, Postgres advisory locks, etc.). The library ships with InMemoryLockManager for single-process deployments.

Implementations should be reentrant within the same execution context (including child tasks spawned by asyncio.gather): if a coroutine already holds the lock for a room and awaits code that tries to acquire the same room lock, the inner acquisition must succeed without deadlocking. This is required because tool handlers (e.g. handoff) may update room state while the inbound pipeline already holds the room lock.

locked abstractmethod async

locked(room_id)

Acquire an exclusive lock for room_id.

Acquisition MUST be cancellation-safe: the inbound pipeline bounds the wait with process_timeout (RFC §13.6), so a caller that gives up queueing is cancelled here. An implementation that leaves the lock taken, or a reference held, on cancellation strands the room — every later event for it queues behind a lock nobody owns.

close async

close()

Release any resources held by the lock manager (e.g. a connection pool). Called by RoomKit.close(). Default is a no-op; overrides MUST be idempotent.

InMemoryLockManager

InMemoryLockManager(max_locks=1024)

Bases: RoomLockManager

In-process per-room asyncio locks with LRU eviction.

Reentrant within the same execution context: if the current context already holds the lock for a given room (including child tasks spawned by asyncio.gather), locked() yields immediately instead of deadlocking.

Suitable for single-process deployments. For multi-process or distributed setups, provide a custom RoomLockManager backed by Redis, Postgres advisory locks, or similar.

size property

size

Return the number of locks currently held.

locked async

locked(room_id)

Acquire the lock for a room (reentrant via ContextVar).

AuthCallback module-attribute

AuthCallback = Callable[[Any], Awaitable[dict[str, Any] | None]]

Async callback for transport authentication.

Receives the connection context (e.g. WebSocket, HTTP request) and returns a metadata dict on success or None to reject the connection.