RoomKit
Minimum version: 0.7.0 | Python 3.12+
RoomKit is a pure async Python library for building multi-channel conversation
systems. It provides room-based abstractions for managing conversations across
SMS, Email, Voice, WebSocket, AI, and other channels with pluggable storage,
identity resolution, hooks, and realtime events.
Pydantic 2.x, fully typed, zero required dependencies beyond Pydantic.
RoomKit is a pure async Python library for building multi-channel conversation systems. It provides a room-based abstraction where conversations happen in rooms, participants communicate through channels, and hooks let you intercept and modify the flow.
Key Concepts
- Room — A container for a conversation. Holds participants, channel bindings, and an ordered event timeline.
- Channel — A communication endpoint (SMS, Email, Voice, AI, WebSocket, etc.). Registered once, attached to many rooms.
- Hook — A function that intercepts events at specific points in the pipeline. Can block, modify, or observe messages.
- Event Router — Broadcasts events to all channels attached to a room, with content transcoding per channel capabilities.
- Identity Pipeline — Maps external sender IDs to known participants with challenge/response flows.
- Realtime Events — Ephemeral events (typing, presence, reactions) that are not stored in history.
Architecture at a Glance
RoomKit uses a hub-and-spoke model. The RoomKit orchestrator sits at the center. Channels connect on the edges. Messages flow inbound through a defined pipeline, get stored, then broadcast outbound to all attached channels.
Inbound Message
-> InboundRoomRouter.route() # Find target room
-> Channel.handle_inbound() # Parse -> RoomEvent
-> IdentityResolver.resolve() # Identify sender
-> BEFORE_BROADCAST hooks # Can block/modify
-> Store event
-> EventRouter.broadcast() # Deliver to all channels
-> Content transcoding # Adapt per channel capabilities
-> Rate limiting + retry
-> AFTER_BROADCAST hooks # Async side effects
What You Can Build
- AI-powered support agents across SMS, WhatsApp, and web chat
- Voice assistants with real-time STT/TTS and interruption handling
- Multi-agent pipelines where specialized agents hand off conversations
- Notification systems that bridge channels (SMS + Email + push)
- Speech-to-speech AI with Gemini Live, OpenAI Realtime, or Grok
Design Principles
- Async-first — All I/O is async. No synchronous blocking.
- Pluggable everything — Storage, identity, routing, AI providers, voice backends — all swappable via ABCs with in-memory defaults.
- Zero required deps — Only
pydantic>=2.9. Everything else is optional extras.
- Type-safe — Strict mypy, full type hints, Pydantic models throughout.
- Python 3.12+ — Uses modern syntax (
X | None, not Optional[X]).
Basic Install
pip install roomkit
# or with uv
uv add roomkit
The core library has a single dependency: pydantic>=2.9.
Optional Extras
RoomKit uses optional extras for provider-specific dependencies. Install only what you need:
AI Providers
pip install roomkit[anthropic] # Anthropic (Claude)
pip install roomkit[openai] # OpenAI (GPT)
pip install roomkit[gemini] # Google Gemini
pip install roomkit[mistral] # Mistral AI
pip install roomkit[vllm] # vLLM local inference (uses openai SDK)
pip install roomkit[azure] # Azure OpenAI
Voice — Speech-to-Text
pip install roomkit[deepgram] # Deepgram STT (cloud)
pip install roomkit[sherpa-onnx] # SherpaOnnx STT (local, offline)
pip install roomkit[gradium] # Gradium STT
pip install roomkit[qwen-asr] # Qwen3 ASR
Voice — Text-to-Speech
pip install roomkit[elevenlabs] # ElevenLabs TTS (cloud)
pip install roomkit[sherpa-onnx] # SherpaOnnx TTS (local, offline)
pip install roomkit[gradium] # Gradium TTS
pip install roomkit[qwen-tts] # Qwen3 TTS
pip install roomkit[neutts] # NeuTTS
Voice — Backends
pip install roomkit[local-audio] # Local mic/speaker (sounddevice + numpy)
pip install roomkit[fastrtc] # FastRTC WebRTC backend
pip install roomkit[rtp] # RTP backend
pip install roomkit[sip] # SIP backend
pip install roomkit[webtransport] # WebTransport backend
Voice — Pipeline
pip install roomkit[webrtc-aec] # WebRTC echo cancellation
pip install roomkit[aicoustics] # ai|coustics denoiser
pip install roomkit[smart-turn] # ML-based turn detection
Realtime Voice (Speech-to-Speech)
pip install roomkit[realtime-openai] # OpenAI Realtime API
pip install roomkit[realtime-gemini] # Google Gemini Live API
Messaging Providers
pip install roomkit[twilio] # Twilio SMS/RCS
pip install roomkit[telegram] # Telegram Bot API
pip install roomkit[teams] # Microsoft Teams (Bot Framework)
pip install roomkit[whatsapp-personal] # WhatsApp Personal (neonize)
pip install roomkit[websocket] # WebSocket source
pip install roomkit[sse] # Server-Sent Events source
Storage & Infrastructure
pip install roomkit[postgres] # PostgreSQL persistence (asyncpg)
pip install roomkit[mcp] # Model Context Protocol tools
pip install roomkit[opentelemetry] # OpenTelemetry tracing
Meta Extras
pip install roomkit[providers] # All AI + transport providers
pip install roomkit[sources] # All event-driven sources
pip install roomkit[dev] # Development (test + lint + type check)
Environment Variables
Provider-specific API keys are passed via configuration objects, not environment variables. Example:
from roomkit.providers.anthropic.config import AnthropicConfig
config = AnthropicConfig(api_key="sk-ant-...")
For voice providers, use lazy loaders to avoid import-time dependency checks:
from roomkit.voice import get_deepgram_provider, get_deepgram_config
DeepgramSTTProvider = get_deepgram_provider()
DeepgramConfig = get_deepgram_config()
stt = DeepgramSTTProvider(DeepgramConfig(api_key="..."))
Development Setup
git clone https://github.com/roomkit-live/roomkit
cd roomkit
uv sync --extra dev # Install all dev dependencies
make all # Run lint + typecheck + security + tests
A complete working example: two WebSocket users chatting with an AI assistant in a moderated room.
from __future__ import annotations
import asyncio
from roomkit import (
ChannelCategory,
HookResult,
HookTrigger,
InboundMessage,
RoomContext,
RoomEvent,
RoomKit,
TextContent,
WebSocketChannel,
)
from roomkit.channels.ai import AIChannel
from roomkit.providers.ai.mock import MockAIProvider
async def main() -> None:
# 1. Create the framework instance
kit = RoomKit()
# 2. Create channels
ws_alice = WebSocketChannel("ws-alice")
ws_bob = WebSocketChannel("ws-bob")
ai = AIChannel("ai-assistant", provider=MockAIProvider(responses=["Got it!"]))
# 3. Register channels with the framework
kit.register_channel(ws_alice)
kit.register_channel(ws_bob)
kit.register_channel(ai)
# 4. Wire up receive callbacks (in production, WebSocket sends to clients)
alice_inbox: list[RoomEvent] = []
bob_inbox: list[RoomEvent] = []
async def alice_recv(_conn: str, event: RoomEvent) -> None:
alice_inbox.append(event)
async def bob_recv(_conn: str, event: RoomEvent) -> None:
bob_inbox.append(event)
ws_alice.register_connection("alice-conn", alice_recv)
ws_bob.register_connection("bob-conn", bob_recv)
# 5. Create a room and attach channels
await kit.create_room(room_id="demo-room")
await kit.attach_channel("demo-room", "ws-alice")
await kit.attach_channel("demo-room", "ws-bob")
await kit.attach_channel(
"demo-room", "ai-assistant", category=ChannelCategory.INTELLIGENCE
)
# 6. Add a BEFORE_BROADCAST hook for content moderation
@kit.hook(HookTrigger.BEFORE_BROADCAST, name="profanity_filter")
async def profanity_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
if isinstance(event.content, TextContent) and "badword" in event.content.body:
return HookResult.block("Message blocked by profanity filter")
return HookResult.allow()
# 7. Send messages through the inbound pipeline
result = await kit.process_inbound(
InboundMessage(
channel_id="ws-alice",
sender_id="alice",
content=TextContent(body="Hello everyone!"),
)
)
print(f"Alice sent 'Hello everyone!' -> blocked={result.blocked}")
# Both Bob and the AI receive Alice's message.
# The AI responds with "Got it!" which is broadcast to Alice and Bob.
# 8. Query stored conversation history
events = await kit.store.list_events("demo-room")
for ev in events:
print(f" [{ev.source.channel_id}] {ev.content.body}")
if __name__ == "__main__":
asyncio.run(main())
What Just Happened
- RoomKit() created the framework with in-memory defaults (store, locks, realtime).
- Channels were registered globally, then attached to a room.
- AI channel was attached with
category=ChannelCategory.INTELLIGENCE — it receives messages and generates responses.
- A BEFORE_BROADCAST hook runs synchronously before every broadcast. It can
block(), allow(), or modify() events.
- process_inbound() ran the full pipeline: route -> parse -> identity -> hooks -> store -> broadcast.
- The AI's response was automatically routed back through the same pipeline (with chain depth tracking to prevent loops).
Core Pattern
Every RoomKit application follows this pattern:
from roomkit import RoomKit
# 1. Create framework
kit = RoomKit()
# 2. Register channels
kit.register_channel(channel)
# 3. Create rooms and attach channels
await kit.create_room(room_id="my-room")
await kit.attach_channel("my-room", "channel-id")
# 4. Process inbound messages
await kit.process_inbound(InboundMessage(...))
Next Steps
- Add hooks for moderation, logging, or analytics (see Hooks)
- Configure AI providers for real LLM responses (see AI Channels)
- Add voice with STT/TTS (see Voice Channels)
- Set up multi-agent orchestration (see Orchestration)
- Deploy with PostgreSQL storage (see Storage)
Hub-and-Spoke Model
RoomKit uses a hub-and-spoke architecture. The RoomKit class is the central hub. Channels are the spokes. Messages flow in through channels, get processed by the hub, then broadcast out to all channels attached to the room.
+------------------+
| RoomKit |
| (orchestrator) |
+--------+---------+
|
+-------------------+-------------------+
| | | | |
+---+---+ +---+---+ +--+---+ +---+---+ +---+---+
| SMS | | Email | | Voice| | AI | | WS |
+-------+ +-------+ +------+ +-------+ +-------+
Inbound Pipeline
Every inbound message follows an immutable pipeline order:
1. InboundRoomRouter.route() # Find target room by channel binding
2. Channel.handle_inbound() # Parse external format -> RoomEvent
3. IdentityResolver.resolve() # Map sender_id -> participant
4. Identity hooks # ON_IDENTITY_AMBIGUOUS / ON_IDENTITY_UNKNOWN
5. Room lock acquired # Per-room atomic processing
6. Idempotency check # Deduplicate by provider_message_id
7. BEFORE_BROADCAST hooks # Sync: can block or modify the event
8. Store event + update counters # Persist to ConversationStore
9. EventRouter.broadcast() # Deliver to all attached channels
-> Content transcoding # Adapt content per channel capabilities
-> Rate limiting # TokenBucketRateLimiter per binding
-> Retry with backoff # RetryPolicy per binding
10. AFTER_BROADCAST hooks # Async: fire-and-forget side effects
11. Room lock released
This order is defined in the RFC (Section 10.1) and must not be reordered.
Channel Categories
Channels have two categories:
- TRANSPORT — Push messages to users (SMS, Email, WebSocket, Voice). Default category.
- INTELLIGENCE — Generate responses (AI, agents). Receives broadcasts, responds through the inbound pipeline.
Channel-to-AI Message Flow (Reentry Loop)
When you send a message from a transport channel, it flows to the AI and back automatically. Here's the complete flow:
1. Transport channel (SMS/WS/Voice) → kit.process_inbound(InboundMessage)
2. Inbound pipeline: route → parse → identity → hooks → store
3. EventRouter.broadcast() → delivers event to ALL attached channels
├── Transport channels: note delivery (no response)
└── AI channel: calls LLM provider → generates response
└── Returns ChannelOutput(response_events=[RoomEvent])
4. REENTRY LOOP: AI response re-enters as new inbound event
├── BEFORE_BROADCAST hooks run again (ConversationRouter stamps routing)
├── Event stored in timeline
├── Broadcast to all channels again
│ ├── Transport channels: DELIVER the AI response to users
│ └── Other AI channels: see the response (may generate follow-up)
└── If follow-up AI responses exist → loop again (chain depth checked)
5. Chain depth limit (default max=5) → stops AI-to-AI infinite loops
6. AFTER_BROADCAST hooks fire (async side effects)
Key insight: The "pipeline" is a reentry loop — not a linear chain. AI responses go back through the same BEFORE_BROADCAST hooks, content transcoding, and broadcast cycle as user messages.
Connecting Channel to AI — Minimal Example
from roomkit import RoomKit, WebSocketChannel, ChannelCategory, InboundMessage, TextContent
from roomkit.channels.ai import AIChannel
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig
kit = RoomKit()
# Transport channel (user-facing)
ws = WebSocketChannel("ws-user")
ws.register_connection("conn-1", on_recv)
kit.register_channel(ws)
# Intelligence channel (AI)
ai = AIChannel("ai", provider=AnthropicAIProvider(AnthropicConfig(
api_key="sk-ant-...", model="claude-sonnet-4-20250514",
)))
kit.register_channel(ai)
# Room wires them together
await kit.create_room(room_id="chat")
await kit.attach_channel("chat", "ws-user") # TRANSPORT (default)
await kit.attach_channel("chat", "ai", category=ChannelCategory.INTELLIGENCE)
# User sends message → AI responds → response delivered to WebSocket
await kit.process_inbound(
InboundMessage(channel_id="ws-user", sender_id="user", content=TextContent(body="Hi!"))
)
# on_recv callback fires with the AI's response
Pipeline vs Pipeline
RoomKit uses "Pipeline" in two contexts — don't confuse them:
| Term | What It Is | Where |
|---|
| Inbound processing pipeline | The 11-step message processing flow (route → parse → identity → hooks → store → broadcast) | core/mixins/inbound_locked.py |
| Pipeline orchestration strategy | A linear agent chain (triage → handler → resolver) for multi-agent handoffs | from roomkit import Pipeline |
The inbound pipeline processes every message. The Pipeline strategy controls which agent handles each turn.
Pluggable Components
Every core component follows the ABC + default pattern:
| Component | ABC | Default | Purpose |
|---|
ConversationStore | store/base.py | InMemoryStore | Room, event, participant persistence |
RoomLockManager | core/locks.py | InMemoryLockManager | Per-room atomic processing |
RealtimeBackend | realtime/base.py | InMemoryRealtime | Ephemeral events (typing, presence) |
IdentityResolver | identity/base.py | None (disabled) | Sender -> participant mapping |
InboundRoomRouter | core/inbound_router.py | DefaultInboundRoomRouter | Route messages to rooms |
Replace any component at construction:
from roomkit import RoomKit
from roomkit.store.postgres import PostgresStore
kit = RoomKit(store=PostgresStore("postgresql://..."))
Room Lifecycle
ACTIVE -> PAUSED -> CLOSED -> ARCHIVED
^ |
+----------+ (can close from paused)
- ACTIVE — Accepting messages, all channels active.
- PAUSED — Messages queued, channels paused. Auto-transition via
inactive_after_seconds.
- CLOSED — No new messages. Auto-transition via
closed_after_seconds.
- ARCHIVED — Terminal state, read-only.
Event Model
Every message becomes a RoomEvent with:
- content — One of 11 content types (TextContent, RichContent, MediaContent, AudioContent, VideoContent, LocationContent, CompositeContent, TemplateContent, EditContent, DeleteContent, SystemContent)
- source — Who sent it (channel_id, participant_id, direction)
- index — Sequential, monotonically increasing per room
- metadata — Arbitrary key-value data
Voice Architecture
The voice subsystem has three layers:
- VoiceBackend — Pure audio transport (mic, SIP, RTP, WebRTC). No speech detection.
- AudioPipeline — Processes audio frames: Resampler -> Recorder -> AEC -> AGC -> Denoiser -> VAD -> Diarization.
- VoiceChannel — Wires backend -> pipeline -> STT/TTS, handles interruption and turn detection.
Inbound: Backend -> [Resampler] -> [Recorder] -> [AEC] -> [AGC] -> [Denoiser] -> VAD -> [Diarization] + [DTMF]
Outbound: TTS -> [PostProcessors] -> [Recorder] -> AEC.feed_reference -> [Resampler] -> Backend
Realtime Voice Architecture
Speech-to-speech AI (Gemini Live, OpenAI Realtime) bypasses STT/TTS entirely:
- RealtimeVoiceProvider — Handles the AI model connection (WebSocket to Gemini/OpenAI).
- RealtimeAudioTransport — Handles browser/client audio (WebSocket or WebRTC).
- RealtimeVoiceChannel — Bridges transport <-> provider with tool calling support.
Creating Rooms
from roomkit import RoomKit
kit = RoomKit()
# Auto-generated ID
room = await kit.create_room()
# Explicit ID with metadata
room = await kit.create_room(
room_id="support-123",
metadata={"topic": "billing", "priority": "high"},
)
Room Lifecycle
from roomkit import RoomKit
kit = RoomKit()
# Create room with auto-pause and auto-close timers
room = await kit.create_room(room_id="session-1")
# Update metadata
await kit.update_room_metadata("session-1", {"status": "escalated"})
# Close a room
await kit.close_room("session-1")
# Check timers (call periodically, e.g. every 60s)
transitioned = await kit.check_all_timers()
Channel Types
RoomKit ships with these channel types:
| Channel | Factory / Class | Category | Use Case |
|---|
| SMS | SMSChannel(id, provider=...) | TRANSPORT | Text messages via Twilio, Telnyx, Sinch |
| RCS | RCSChannel(id, provider=...) | TRANSPORT | Rich messaging via Twilio, Telnyx |
| Email | EmailChannel(id, provider=...) | TRANSPORT | Email via ElasticEmail, SendGrid |
| WhatsApp | WhatsAppChannel(id, provider=...) | TRANSPORT | WhatsApp Business API |
| WhatsApp Personal | WhatsAppPersonalChannel(id, provider=...) | TRANSPORT | WhatsApp via neonize |
| Messenger | MessengerChannel(id, provider=...) | TRANSPORT | Facebook Messenger |
| Telegram | TelegramChannel(id, provider=...) | TRANSPORT | Telegram Bot API |
| Teams | TeamsChannel(id, provider=...) | TRANSPORT | Microsoft Teams Bot Framework |
| HTTP | HTTPChannel(id, provider=...) | TRANSPORT | Generic webhook |
| WebSocket | WebSocketChannel(id) | TRANSPORT | Real-time bidirectional |
| AI | AIChannel(id, provider=...) | INTELLIGENCE | LLM responses |
| Agent | Agent(id, provider=...) | INTELLIGENCE | Agent with tools + greeting |
| Voice | VoiceChannel(id, stt=..., tts=..., backend=...) | TRANSPORT | Real-time audio |
| Realtime Voice | RealtimeVoiceChannel(id, provider=...) | INTELLIGENCE | Speech-to-speech AI |
Registering and Attaching Channels
Channels are registered globally, then attached to specific rooms:
from roomkit import RoomKit, SMSChannel, ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig
kit = RoomKit()
# Create and register channels
sms = SMSChannel("sms-main", provider=TwilioSMSProvider(TwilioConfig(
account_sid="AC...",
auth_token="...",
from_number="+1234567890",
)))
ai = AIChannel("ai-agent", provider=AnthropicAIProvider(AnthropicConfig(
api_key="sk-ant-...",
model="claude-sonnet-4-20250514",
)))
kit.register_channel(sms)
kit.register_channel(ai)
# Create room and attach
await kit.create_room(room_id="support-room")
await kit.attach_channel("support-room", "sms-main")
await kit.attach_channel("support-room", "ai-agent", category=ChannelCategory.INTELLIGENCE)
Channel Access Levels
Control what a channel can do in a room:
from roomkit import Access
# Default: read and write
await kit.attach_channel("room", "channel", access=Access.READ_WRITE)
# Read only: receives messages but cannot send
await kit.attach_channel("room", "channel", access=Access.READ_ONLY)
# Write only: sends messages but doesn't receive broadcasts
await kit.attach_channel("room", "channel", access=Access.WRITE_ONLY)
# None: temporarily disabled
await kit.set_access("room", "channel", Access.NONE)
Muting Channels
Muting suppresses a channel's output (AI responses) without detaching it. Side effects (tasks, observations) still fire.
# Mute AI output
await kit.mute("room", "ai-agent")
# Unmute
await kit.unmute("room", "ai-agent")
# Mute only the output direction (AI won't respond, but still sees messages)
await kit.mute_output("room", "ai-agent")
await kit.unmute_output("room", "ai-agent")
Per-Room Channel Configuration
Pass metadata when attaching to customize per-room behavior:
await kit.attach_channel(
"weather-room",
"ai-agent",
category=ChannelCategory.INTELLIGENCE,
metadata={
"system_prompt": "You are a weather assistant.",
"temperature": 0.3,
"tools": [
{
"name": "get_weather",
"description": "Get current weather for a city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string"},
},
"required": ["city"],
},
},
],
},
)
Content Types
Events carry typed content. RoomKit supports 11 content types, all discriminated unions on the type field:
| Type | type Literal | Key Fields | Use Case |
|---|
TextContent | "text" | body, language | Plain text messages |
RichContent | "rich" | body, format (html/markdown), plain_text, buttons, cards, quick_replies | Formatted messages with UI elements |
MediaContent | "media" | url, mime_type, filename, size_bytes, caption | Images, documents, files |
AudioContent | "audio" | url, mime_type, duration_seconds, transcript | Voice messages |
VideoContent | "video" | url, mime_type, duration_seconds, thumbnail_url | Video messages |
LocationContent | "location" | latitude, longitude, label, address | Geographic coordinates |
CompositeContent | "composite" | parts (list of EventContent, max depth 5) | Multi-part messages |
TemplateContent | "template" | template_id, language, parameters, body | WhatsApp Business / RCS templates |
SystemContent | "system" | body, code, data (dict) | System-generated events |
EditContent | "edit" | target_event_id, new_content, edit_source | Message edits |
DeleteContent | "delete" | target_event_id, delete_type, reason | Message deletion |
from roomkit.models.event import (
TextContent,
RichContent,
MediaContent,
AudioContent,
VideoContent,
LocationContent,
CompositeContent,
TemplateContent,
SystemContent,
EditContent,
DeleteContent,
)
# Plain text
TextContent(body="Hello!")
# Rich content with buttons and Markdown
RichContent(
body="Choose an option:",
format="markdown",
plain_text="Choose an option:",
buttons=[{"text": "Option A", "payload": "a"}, {"text": "Option B", "payload": "b"}],
)
# Media attachment
MediaContent(url="https://example.com/image.png", mime_type="image/png", caption="Product photo")
# Audio with transcript
AudioContent(url="https://example.com/voice.ogg", mime_type="audio/ogg", transcript="Hello there")
# Location
LocationContent(latitude=40.7128, longitude=-74.0060, label="NYC Office", address="New York, NY")
# Template (WhatsApp Business)
TemplateContent(template_id="order_confirmation", language="en", parameters={"1": "ORD-123"})
# Multi-part message (max depth 5)
CompositeContent(parts=[
TextContent(body="Here's the photo:"),
MediaContent(url="https://example.com/photo.jpg", mime_type="image/jpeg"),
])
# Edit a previous message
EditContent(target_event_id="evt-abc", new_content=TextContent(body="Corrected text"))
# Delete a message
DeleteContent(target_event_id="evt-abc", delete_type="sender", reason="User retracted")
Content Transcoding
When broadcasting events, the EventRouter automatically transcodes content to match each target channel's capabilities. A channel that only supports TEXT will receive a text fallback of a RichContent message.
How Transcoding Works
- Event broadcast starts → EventRouter iterates over all target channel bindings
- For each target, call
ContentTranscoder.transcode(content, source_binding, target_binding)
- Transcoder checks
target_binding.capabilities.media_types to decide if content passes through or needs conversion
- If transcode returns
None → delivery is skipped for that channel (content cannot be represented)
Channel Capabilities
Each channel binding declares what content types it supports via ChannelCapabilities:
from roomkit.models.channel import ChannelCapabilities
from roomkit.models.enums import ChannelMediaType
# SMS channel: text only, 160 char limit
sms_caps = ChannelCapabilities(
media_types=[ChannelMediaType.TEXT],
max_length=160,
)
# WebSocket channel: rich content, media, audio, video
ws_caps = ChannelCapabilities(
media_types=[
ChannelMediaType.TEXT,
ChannelMediaType.RICH,
ChannelMediaType.MEDIA,
ChannelMediaType.AUDIO,
ChannelMediaType.VIDEO,
ChannelMediaType.LOCATION,
],
supports_edit=True,
supports_delete=True,
supports_reactions=True,
supports_typing=True,
)
ChannelMediaType enum values: TEXT, RICH, MEDIA, AUDIO, VIDEO, LOCATION, TEMPLATE.
Default Fallback Chain
The DefaultContentTranscoder applies these fallback rules:
| Content Type | If Target Supports It | Fallback |
|---|
TextContent | Always passes through | — |
RichContent | RICH in media_types → pass | TextContent(plain_text or body) |
MediaContent | MEDIA in media_types → pass | TextContent("[Media: {caption or filename or url}]") |
AudioContent | AUDIO in media_types → pass | TextContent(transcript) or "[Voice message: {url}]" |
VideoContent | VIDEO in media_types → pass | TextContent("[Video: {url}]") |
LocationContent | LOCATION in media_types → pass | TextContent("[Location: {label} ({lat}, {lon})]") |
CompositeContent | Recursive transcode of all parts | If all parts become text → flatten to single TextContent |
TemplateContent | TEMPLATE in media_types → pass | TextContent(body or "[Template: {id}]") |
EditContent | supports_edit → pass | Transcode new_content + prefix "Correction:" |
DeleteContent | supports_delete → pass | TextContent("[Message deleted]") |
Custom Content Transcoder
Implement the ContentTranscoder ABC to customize how content is adapted:
from roomkit.core.router import ContentTranscoder
from roomkit.models.channel import ChannelBinding
from roomkit.models.event import EventContent, TextContent, MediaContent
from roomkit.models.enums import ChannelMediaType
class MyTranscoder(ContentTranscoder):
async def transcode(
self,
content: EventContent,
source_binding: ChannelBinding,
target_binding: ChannelBinding,
) -> EventContent | None:
target_types = target_binding.capabilities.media_types
# Custom: convert images to descriptive text for voice channels
if isinstance(content, MediaContent) and content.mime_type.startswith("image/"):
if ChannelMediaType.MEDIA not in target_types:
caption = content.caption or "an image"
return TextContent(body=f"[Image received: {caption}]")
# Custom: truncate long text for SMS
if isinstance(content, TextContent):
max_len = target_binding.capabilities.max_length
if max_len and len(content.body) > max_len:
return TextContent(body=content.body[: max_len - 3] + "...")
# Fall through to default behavior for other types
return content
# Override the default transcoder on the RoomKit instance
kit = RoomKit()
kit._transcoder = MyTranscoder()
Multichannel Transcoding Example
A single event gets adapted differently for each channel:
# User sends a rich message with an image from WebSocket
content = CompositeContent(parts=[
RichContent(body="**Check this out!**", format="markdown", plain_text="Check this out!"),
MediaContent(url="https://example.com/photo.jpg", mime_type="image/jpeg", caption="Sunset"),
])
# WebSocket channel: receives CompositeContent as-is (supports RICH + MEDIA)
# SMS channel: receives TextContent("Check this out!\n[Media: Sunset]")
# Voice channel (TTS): receives TextContent("Check this out!\n[Media: Sunset]")
Detaching Channels
await kit.detach_channel("room", "channel-id")
Binding Metadata Updates
await kit.update_binding_metadata("room", "ai-agent", {"temperature": 0.9})
Querying Rooms
# Get a room
room = await kit.get_room("room-id")
# List bindings
bindings = await kit.store.list_bindings("room-id")
# List participants
participants = await kit.store.list_participants("room-id")
# Query event timeline
events = await kit.get_timeline("room-id", offset=0, limit=50)
Hooks intercept events at specific points in the processing pipeline. They can block messages, modify content, trigger side effects, or observe the conversation.
Hook Basics
from roomkit import RoomKit, HookTrigger, HookExecution, HookResult, RoomEvent, RoomContext
kit = RoomKit()
# Sync hook: runs BEFORE broadcast, can block or modify
@kit.hook(HookTrigger.BEFORE_BROADCAST)
async def content_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
if "spam" in event.content.body.lower():
return HookResult.block("Spam detected")
return HookResult.allow()
# Async hook: runs AFTER broadcast, fire-and-forget
@kit.hook(HookTrigger.AFTER_BROADCAST, execution=HookExecution.ASYNC)
async def log_event(event: RoomEvent, ctx: RoomContext) -> None:
await analytics.track("message", {"room": event.room_id})
HookResult
Sync hooks (BEFORE_BROADCAST) must return a HookResult:
from roomkit import HookResult, TextContent
# Allow the event to proceed
HookResult.allow()
# Block the event with a reason
HookResult.block("Contains prohibited content")
# Modify the event before broadcast
modified = event.model_copy(update={"content": TextContent(body="[REDACTED]")})
HookResult.modify(modified)
Hook Priority
Lower priority numbers run first. Default is 0.
# Runs first (priority=0)
@kit.hook(HookTrigger.BEFORE_BROADCAST, name="profanity_filter", priority=0)
async def profanity_filter(event: RoomEvent, ctx: RoomContext) -> HookResult:
blocked_words = {"badword", "spam", "scam"}
if isinstance(event.content, TextContent):
words = set(event.content.body.lower().split())
if words & blocked_words:
return HookResult.block(f"Blocked: {words & blocked_words}")
return HookResult.allow()
# Runs second (priority=1)
@kit.hook(HookTrigger.BEFORE_BROADCAST, name="pii_redactor", priority=1)
async def pii_redactor(event: RoomEvent, ctx: RoomContext) -> HookResult:
import re
if isinstance(event.content, TextContent):
redacted = re.sub(
r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b", "[REDACTED]", event.content.body
)
if redacted != event.content.body:
modified = event.model_copy(update={"content": TextContent(body=redacted)})
return HookResult.modify(modified)
return HookResult.allow()
Hook Filters
Filter hooks by channel type, channel ID, or direction:
from roomkit.models.enums import ChannelType, ChannelDirection
@kit.hook(
HookTrigger.AFTER_BROADCAST,
execution=HookExecution.ASYNC,
channel_types={ChannelType.SMS},
directions={ChannelDirection.INBOUND},
priority=10,
)
async def sms_audit(event: RoomEvent, ctx: RoomContext) -> None:
await audit_log.record(event)
Room-Scoped Hooks
Add hooks to specific rooms instead of globally:
from roomkit import HookExecution
await kit.add_room_hook(
room_id="vip-room",
trigger=HookTrigger.BEFORE_BROADCAST,
execution=HookExecution.SYNC,
fn=my_hook_function,
name="vip_filter",
)
# Remove later
await kit.remove_room_hook("vip-room", "vip_filter")
Complete Hook Trigger Reference
Event Pipeline
| Trigger | Execution | Signature | Description |
|---|
BEFORE_BROADCAST | SYNC | (event, ctx) -> HookResult | Before event is stored and broadcast. Can block/modify. |
AFTER_BROADCAST | ASYNC | (event, ctx) -> None | After event is broadcast. Fire-and-forget side effects. |
Channel Lifecycle
| Trigger | Execution | Signature | Description |
|---|
ON_CHANNEL_ATTACHED | ASYNC | (event, ctx) -> None | Channel was attached to a room |
ON_CHANNEL_DETACHED | ASYNC | (event, ctx) -> None | Channel was detached from a room |
ON_CHANNEL_MUTED | ASYNC | (event, ctx) -> None | Channel was muted in a room |
ON_CHANNEL_UNMUTED | ASYNC | (event, ctx) -> None | Channel was unmuted in a room |
Room Lifecycle
| Trigger | Execution | Signature | Description |
|---|
ON_ROOM_CREATED | ASYNC | (event, ctx) -> None | Room was created |
ON_ROOM_PAUSED | ASYNC | (event, ctx) -> None | Room was paused |
ON_ROOM_CLOSED | ASYNC | (event, ctx) -> None | Room was closed |
Identity
| Trigger | Execution | Signature | Description |
|---|
ON_IDENTITY_AMBIGUOUS | SYNC | (event, ctx) -> IdentityHookResult | Multiple identity matches found |
ON_IDENTITY_UNKNOWN | SYNC | (event, ctx) -> IdentityHookResult | No identity match found |
ON_PARTICIPANT_IDENTIFIED | ASYNC | (event, ctx) -> None | Participant was successfully identified |
Delivery
| Trigger | Execution | Signature | Description |
|---|
ON_DELIVERY_STATUS | ASYNC | (status) -> None | Delivery status update from provider |
Side Effects
| Trigger | Execution | Signature | Description |
|---|
ON_TASK_CREATED | ASYNC | (event, ctx) -> None | AI extracted a task from conversation |
ON_ERROR | ASYNC | (event, ctx) -> None | Error occurred during processing |
Voice
| Trigger | Execution | Signature | Description |
|---|
ON_SPEECH_START | ASYNC | (event, ctx) -> None | VAD detected speech start |
ON_SPEECH_END | ASYNC | (event, ctx) -> None | VAD detected speech end |
ON_TRANSCRIPTION | ASYNC | (event, ctx) -> None | STT produced a transcription |
BEFORE_TTS | SYNC | (event, ctx) -> HookResult | Before text is sent to TTS. Can block/modify. |
AFTER_TTS | ASYNC | (event, ctx) -> None | After TTS audio is generated |
Voice Pipeline
| Trigger | Execution | Signature | Description |
|---|
ON_VAD_SILENCE | ASYNC | (event, ctx) -> None | VAD detected silence |
ON_VAD_AUDIO_LEVEL | ASYNC | (event, ctx) -> None | Audio level update from VAD |
ON_SPEAKER_CHANGE | ASYNC | (event, ctx) -> None | Diarization detected speaker change |
ON_BARGE_IN | ASYNC | (event, ctx) -> None | User interrupted TTS playback |
ON_TTS_CANCELLED | ASYNC | (event, ctx) -> None | TTS playback was cancelled |
ON_DTMF | ASYNC | (event, ctx) -> None | DTMF tone detected |
ON_TURN_COMPLETE | ASYNC | (event, ctx) -> None | Turn detector says turn is complete |
ON_TURN_INCOMPLETE | ASYNC | (event, ctx) -> None | Turn detector says turn is incomplete |
ON_BACKCHANNEL | ASYNC | (event, ctx) -> None | Backchannel detected (uh-huh, yeah) |
ON_RECORDING_STARTED | ASYNC | (event, ctx) -> None | Audio recording started |
ON_RECORDING_STOPPED | ASYNC | (event, ctx) -> None | Audio recording stopped |
ON_INPUT_AUDIO_LEVEL | ASYNC | (event, ctx) -> None | Input audio level update |
ON_OUTPUT_AUDIO_LEVEL | ASYNC | (event, ctx) -> None | Output audio level update |
Tool Execution
| Trigger | Execution | Signature | Description |
|---|
ON_TOOL_CALL | ASYNC | (event, ctx) -> None | AI invoked a tool (fires from AIChannel and RealtimeVoiceChannel) |
Realtime Voice
| Trigger | Execution | Signature | Description |
|---|
ON_REALTIME_TEXT_INJECTED | ASYNC | (event, ctx) -> None | Text was injected into realtime session |
Observability
| Trigger | Execution | Signature | Description |
|---|
ON_FEEDBACK | ASYNC | (Observation, ctx) -> None | User submitted quality feedback via kit.submit_feedback() |
Framework Events
Framework events are lightweight lifecycle notifications (not message events):
@kit.on("room_created")
async def on_room_created(event):
print(f"Room created: {event.data['room_id']}")
@kit.on("voice_session_started")
async def on_voice(event):
print(f"Voice session: {event.data['session_id']}")
Available framework event types: room_created, room_closed, room_paused, room_channel_attached, room_channel_detached, channel_connected, channel_disconnected, voice_session_started, voice_session_ended, source_attached, source_detached, source_error, source_exhausted.
AIChannel connects rooms to LLM providers. When a message is broadcast to an AI channel, it generates a response using conversation history and re-enters it through the inbound pipeline.
Basic Setup
from roomkit import RoomKit, ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig
kit = RoomKit()
ai = AIChannel(
"ai-assistant",
provider=AnthropicAIProvider(AnthropicConfig(
api_key="sk-ant-...",
model="claude-sonnet-4-20250514",
)),
system_prompt="You are a helpful customer support agent.",
temperature=0.7,
)
kit.register_channel(ai)
await kit.create_room(room_id="support")
await kit.attach_channel("support", "ai-assistant", category=ChannelCategory.INTELLIGENCE)
AI Providers
| Provider | Class | Config | Extra |
|---|
| Anthropic (Claude) | AnthropicAIProvider | AnthropicConfig | roomkit[anthropic] |
| OpenAI (GPT) | OpenAIAIProvider | OpenAIConfig | roomkit[openai] |
| Google Gemini | GeminiAIProvider | GeminiConfig | roomkit[gemini] |
| Mistral | MistralAIProvider | MistralConfig | roomkit[mistral] |
| Azure OpenAI | AzureAIProvider | AzureAIConfig | roomkit[azure] |
| vLLM (local) | create_vllm_provider() | VLLMConfig | roomkit[vllm] |
| Mock (testing) | MockAIProvider | — | built-in |
# OpenAI
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig
provider = OpenAIAIProvider(OpenAIConfig(api_key="sk-...", model="gpt-4o"))
# Gemini
from roomkit.providers.gemini.ai import GeminiAIProvider
from roomkit.providers.gemini.config import GeminiConfig
provider = GeminiAIProvider(GeminiConfig(api_key="...", model="gemini-2.0-flash"))
# Mock (for testing)
from roomkit.providers.ai.mock import MockAIProvider
provider = MockAIProvider(responses=["Hello!", "How can I help?"])
Agent Class
Agent extends AIChannel with role, description, greeting, and memory support — designed for multi-agent orchestration:
from roomkit import Agent
from roomkit.providers.ai.mock import MockAIProvider
agent = Agent(
"support-agent",
provider=MockAIProvider(responses=["I can help with that."]),
role="Customer support specialist",
description="Handles billing and account questions",
system_prompt="You are a support specialist. Be concise and helpful.",
greeting="Hi! How can I help you today?",
)
Tool Calling
Define tools as JSON schema and attach them to the AI channel:
from roomkit import ChannelCategory
from roomkit.channels.ai import AIChannel
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig
ai = AIChannel(
"ai-assistant",
provider=OpenAIAIProvider(OpenAIConfig(api_key="sk-...", model="gpt-4o")),
system_prompt="You help users check the weather.",
tools=[
{
"name": "get_weather",
"description": "Get current weather for a city",
"parameters": {
"type": "object",
"properties": {
"city": {"type": "string", "description": "City name"},
"units": {"type": "string", "enum": ["celsius", "fahrenheit"]},
},
"required": ["city"],
},
},
],
)
Tool Handler
Register a handler to execute tool calls via the constructor:
async def handle_tools(name: str, arguments: dict) -> str:
if name == "get_weather":
city = arguments["city"]
return f'{{"temperature": 22, "condition": "sunny", "city": "{city}"}}'
return '{"error": "Unknown tool"}'
ai = AIChannel(
"ai-assistant",
provider=provider,
tools=[...],
tool_handler=handle_tools,
)
Tool Protocol (Tool ABC)
For structured tool definitions, use the Tool base class:
from roomkit.tools.base import Tool
class GetWeather(Tool):
name = "get_weather"
description = "Get current weather for a city"
parameters = {
"type": "object",
"properties": {
"city": {"type": "string"},
},
"required": ["city"],
}
async def execute(self, arguments: dict) -> str:
return '{"temperature": 22, "condition": "sunny"}'
ai = AIChannel("ai", provider=provider, tools=[GetWeather()])
MCP Tool Provider
Integrate Model Context Protocol servers:
from roomkit.tools.mcp import MCPToolProvider
mcp = MCPToolProvider(server_command=["uvx", "mcp-server-sqlite", "--db", "data.db"])
await mcp.initialize()
ai = AIChannel("ai", provider=provider, tools=mcp.tools())
Per-Room Configuration
Override AI settings per room via binding metadata:
await kit.attach_channel(
"billing-room",
"ai-agent",
category=ChannelCategory.INTELLIGENCE,
metadata={
"system_prompt": "You are a billing specialist.",
"temperature": 0.3,
"tools": [...],
},
)
Streaming
AIChannel supports streaming responses to WebSocket clients:
from roomkit import WebSocketChannel
ws = WebSocketChannel("ws-user")
# Register with stream support
ws.register_connection("conn-1", on_recv, stream_send_fn=on_stream)
async def on_stream(conn_id: str, msg) -> None:
# StreamStart, StreamChunk, StreamEnd
print(f"Stream: {msg}")
AI Thinking/Reasoning
Some providers support extended thinking:
ai = AIChannel(
"ai",
provider=AnthropicAIProvider(AnthropicConfig(
api_key="...",
model="claude-sonnet-4-20250514",
)),
system_prompt="Think step by step.",
thinking_budget=4096, # Setting a budget enables thinking mode
)
Vision Support
AI providers that support vision can process images sent as MediaContent:
from roomkit.models.event import MediaContent
await kit.process_inbound(
InboundMessage(
channel_id="ws-user",
sender_id="user",
content=MediaContent(url="https://example.com/chart.png", mime_type="image/png"),
)
)
# AI sees the image and responds with analysis
Tool Call Events
AIChannel automatically broadcasts ephemeral TOOL_CALL_START and TOOL_CALL_END events when executing tools. Subscribe to these for UI indicators:
await kit.subscribe_room("room-1", my_callback)
# Callback receives:
# TOOL_CALL_START: {tool_calls: [{id, name, arguments}], round, channel_id}
# TOOL_CALL_END: {tool_calls: [{id, name, result}], round, channel_id, duration_ms}
VoiceChannel handles real-time audio conversations with speech-to-text, text-to-speech, and an audio processing pipeline.
Basic Voice Setup
from roomkit import RoomKit, VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, MockVADProvider, VADEvent, VADEventType
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.backends.mock import MockVoiceBackend
kit = RoomKit()
# Create providers
backend = MockVoiceBackend()
stt = MockSTTProvider(transcriptions=["Hello, how can I help?"])
tts = MockTTSProvider()
vad = MockVADProvider(events=[
VADEvent(type=VADEventType.SPEECH_START),
None,
VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"audio"),
])
# Create voice channel with pipeline
voice = VoiceChannel(
"voice-agent",
stt=stt,
tts=tts,
backend=backend,
pipeline=AudioPipelineConfig(vad=vad),
)
kit.register_channel(voice)
Joining a Voice Session
# Create room and attach voice channel
await kit.create_room(room_id="call-room")
await kit.attach_channel("call-room", "voice-agent")
# Join a participant to the voice session
session = await kit.join(
room_id="call-room",
channel_id="voice-agent",
participant_id="caller-1",
)
# Leave when done
await kit.leave(session)
STT Providers
| Provider | Class | Config | Extra |
|---|
| Deepgram | DeepgramSTTProvider | DeepgramConfig | roomkit[deepgram] |
| SherpaOnnx | SherpaOnnxSTTProvider | SherpaOnnxSTTConfig | roomkit[sherpa-onnx] |
| Gradium | GradiumSTTProvider | GradiumSTTConfig | roomkit[gradium] |
| Qwen3 ASR | Qwen3ASRProvider | Qwen3ASRConfig | roomkit[qwen-asr] |
| Mock | MockSTTProvider | — | built-in |
Use lazy loaders to avoid import-time dependency checks:
from roomkit.voice import get_deepgram_provider, get_deepgram_config
DeepgramSTTProvider = get_deepgram_provider()
DeepgramConfig = get_deepgram_config()
stt = DeepgramSTTProvider(DeepgramConfig(
api_key="...",
model="nova-2",
language="en",
))
TTS Providers
| Provider | Class | Config | Extra |
|---|
| ElevenLabs | ElevenLabsTTSProvider | ElevenLabsConfig | roomkit[elevenlabs] |
| SherpaOnnx | SherpaOnnxTTSProvider | SherpaOnnxTTSConfig | roomkit[sherpa-onnx] |
| Gradium | GradiumTTSProvider | GradiumTTSConfig | roomkit[gradium] |
| Qwen3 | Qwen3TTSProvider | Qwen3TTSConfig | roomkit[qwen-tts] |
| NeuTTS | NeuTTSProvider | NeuTTSConfig | roomkit[neutts] |
| Grok TTS | GrokTTSProvider | GrokTTSConfig | xAI |
| Mock | MockTTSProvider | — | built-in |
from roomkit.voice import get_elevenlabs_provider, get_elevenlabs_config
ElevenLabsTTSProvider = get_elevenlabs_provider()
ElevenLabsConfig = get_elevenlabs_config()
tts = ElevenLabsTTSProvider(ElevenLabsConfig(
api_key="...",
voice_id="21m00Tcm4TlvDq8ikWAM",
model="eleven_turbo_v2",
))
Voice Backends
Backends handle audio transport between the framework and participants:
| Backend | Class | Extra | Use Case |
|---|
| Local mic/speaker | LocalAudioBackend | roomkit[local-audio] | Development/testing |
| FastRTC (WebRTC) | FastRTCVoiceBackend | roomkit[fastrtc] | Browser-based voice |
| RTP | RTPVoiceBackend | roomkit[rtp] | VoIP integration |
| SIP | SIPVoiceBackend | roomkit[sip] | Telephony |
| WebTransport | WebTransportBackend | roomkit[webtransport] | Low-latency web |
| Mock | MockVoiceBackend | built-in | Testing |
from roomkit.voice import get_local_audio_backend
LocalAudioBackend = get_local_audio_backend()
backend = LocalAudioBackend(sample_rate=16000, channels=1)
Interruption Handling
Four strategies for handling user speech during TTS playback:
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
voice = VoiceChannel(
"voice",
stt=stt,
tts=tts,
backend=backend,
pipeline=AudioPipelineConfig(vad=vad),
interruption=InterruptionConfig(
strategy=InterruptionStrategy.CONFIRMED,
min_speech_ms=300, # Wait 300ms of sustained speech before interrupting
),
)
| Strategy | Behavior |
|---|
IMMEDIATE | Interrupt on any detected speech |
CONFIRMED | Wait for sustained speech (min_speech_ms). Default. |
SEMANTIC | Use BackchannelDetector to ignore "uh-huh", "yeah" |
DISABLED | Never interrupt TTS playback |
Voice Greeting
Send a greeting when a session starts:
await kit.send_greeting(
room_id="call-room",
channel_id="voice-agent",
greeting="Welcome! How can I help you today?",
session=session,
)
Or configure on the Agent:
from roomkit import Agent
agent = Agent(
"voice-agent",
provider=provider,
greeting="Welcome! How can I help you today?",
stt=stt,
tts=tts,
backend=backend,
pipeline=AudioPipelineConfig(vad=vad),
)
Voice Hooks
from roomkit import HookTrigger, HookExecution
@kit.hook(HookTrigger.ON_SPEECH_START, execution=HookExecution.ASYNC)
async def on_speech(event, ctx):
print("User started speaking")
@kit.hook(HookTrigger.ON_TRANSCRIPTION, execution=HookExecution.ASYNC)
async def on_transcription(event, ctx):
print(f"Transcription: {event.content.body}")
@kit.hook(HookTrigger.BEFORE_TTS)
async def before_tts(event, ctx):
# Can modify or block TTS text
return HookResult.allow()
@kit.hook(HookTrigger.ON_BARGE_IN, execution=HookExecution.ASYNC)
async def on_barge_in(event, ctx):
print("User interrupted the AI")
Audio Bridging
Bridge audio between sessions for human-to-human voice calls:
voice = VoiceChannel("voice", backend=backend, bridge=True)
# With bridge + STT for live transcription
voice = VoiceChannel("voice", stt=stt, backend=backend, bridge=True)
Audio bridge supports N-party calls with mixing and cross-rate resampling.
DTMF
Send and detect DTMF tones:
# Send DTMF
await voice.send_dtmf(session, digit="1", duration_ms=160)
# Detect DTMF via hook
@kit.hook(HookTrigger.ON_DTMF, execution=HookExecution.ASYNC)
async def on_dtmf(event, ctx):
print(f"DTMF digit: {event.data['digit']}")
The audio pipeline sits between the voice backend and STT/TTS, processing audio through pluggable stages.
Pipeline Architecture
Inbound: Backend -> [Resampler] -> [Recorder] -> [AEC] -> [AGC] -> [Denoiser] -> VAD -> [Diarization] + [DTMF]
Outbound: TTS -> [PostProcessors] -> [Recorder] -> AEC.feed_reference -> [Resampler] -> Backend
All stages are optional except VAD (required for speech detection). Stages in brackets are skipped if not configured.
Pipeline Configuration
from roomkit import VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, VADConfig
pipeline = AudioPipelineConfig(
resampler=my_resampler, # Sample rate conversion
vad=my_vad_provider, # Voice activity detection (required)
denoiser=my_denoiser, # Background noise removal
diarization=my_diarizer, # Speaker identification
aec=my_aec, # Echo cancellation
agc=my_agc, # Automatic gain control
dtmf=my_dtmf_detector, # DTMF tone detection
recorder=my_recorder, # Audio recording
recording_config=my_rec_config, # Recording settings
turn_detector=my_turn_detector, # Turn-taking detection
vad_config=VADConfig(
silence_threshold_ms=500, # Silence before speech_end
),
)
voice = VoiceChannel(
"voice",
stt=stt,
tts=tts,
backend=backend,
pipeline=pipeline,
)
Capability-Aware Skipping
AEC and AGC stages automatically skip when the backend declares native capabilities:
from roomkit.voice.base import VoiceCapability
# If backend has NATIVE_AEC, pipeline skips the AEC stage
# If backend has NATIVE_AGC, pipeline skips the AGC stage
Pipeline Stages Reference
Resampler
Converts audio between sample rates (e.g., 8kHz SIP to 16kHz STT).
from roomkit.voice.pipeline.resampler import LinearResamplerProvider
resampler = LinearResamplerProvider()
# Or use SincResampler for higher quality
VAD (Voice Activity Detection)
Detects speech start/end events. Required for the pipeline.
from roomkit.voice.pipeline import MockVADProvider, VADEvent, VADEventType, VADConfig
# Mock for testing
vad = MockVADProvider(events=[
VADEvent(type=VADEventType.SPEECH_START),
None, # No event for this frame
VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech-data"),
])
# Production: SherpaOnnx VAD (local, offline)
from roomkit.voice import get_sherpa_onnx_vad_provider, get_sherpa_onnx_vad_config
SherpaVAD = get_sherpa_onnx_vad_provider()
SherpaVADConfig = get_sherpa_onnx_vad_config()
vad = SherpaVAD(SherpaVADConfig(threshold=0.5))
AEC (Acoustic Echo Cancellation)
Removes echo from the microphone signal caused by speaker output.
# Speex AEC
from roomkit.voice import get_speex_aec_provider
SpeexAEC = get_speex_aec_provider()
aec = SpeexAEC(sample_rate=16000, frame_size=160, filter_length=1024)
# WebRTC AEC
# pip install roomkit[webrtc-aec]
The pipeline feeds TTS audio as reference to the AEC via process_outbound().
AGC (Automatic Gain Control)
Normalizes audio volume levels.
from roomkit.voice.pipeline.agc import AGCConfig
from roomkit.voice.pipeline.agc.mock import MockAGCProvider
agc = MockAGCProvider()
Denoiser
Removes background noise from audio.
# RNNoise (local, CPU-based)
from roomkit.voice import get_rnnoise_denoiser_provider
RNNoise = get_rnnoise_denoiser_provider()
denoiser = RNNoise()
# ai|coustics Quail (cloud API)
# pip install roomkit[aicoustics]
# SherpaOnnx denoiser (local ONNX model)
from roomkit.voice import get_sherpa_onnx_denoiser_provider, get_sherpa_onnx_denoiser_config
SherpaDenoiser = get_sherpa_onnx_denoiser_provider()
SherpaDenoiserConfig = get_sherpa_onnx_denoiser_config()
denoiser = SherpaDenoiser(SherpaDenoiserConfig())
Diarization
Identifies different speakers in multi-speaker audio.
from roomkit.voice.pipeline.diarization.mock import MockDiarizationProvider
diarizer = MockDiarizationProvider()
DTMF Detection
Detects dual-tone multi-frequency signals (phone keypad tones).
from roomkit.voice.pipeline.dtmf.mock import MockDTMFDetector
dtmf = MockDTMFDetector()
DTMF runs in parallel with other pipeline stages (before AEC/AGC/denoiser).
Audio Recorder
Records inbound and outbound audio.
from roomkit.voice.pipeline.recorder.mock import MockAudioRecorder
from roomkit.voice.pipeline.recorder import RecordingConfig
recorder = MockAudioRecorder()
config = RecordingConfig(
format="wav",
sample_rate=16000,
channels=1,
)
Turn Detector
Determines when a speaker's turn is complete (post-STT):
from roomkit.voice.pipeline.turn.mock import MockTurnDetector
turn = MockTurnDetector()
# Production: Smart turn detection (ML-based)
# pip install roomkit[smart-turn]
Turn detection accumulates transcription fragments until is_complete=True, then routes the combined text.
Backchannel Detector
Classifies short utterances as backchannel (e.g., "uh-huh", "yeah") to prevent false interruptions:
from roomkit.voice.pipeline.backchannel.mock import MockBackchannelDetector
bc = MockBackchannelDetector()
Used with InterruptionStrategy.SEMANTIC.
Post-Processors
Custom audio transformations on outbound TTS audio:
from roomkit.voice.pipeline.postprocessor.base import AudioPostProcessor
Interruption Handling
The InterruptionHandler manages what happens when the user speaks during TTS playback:
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
config = InterruptionConfig(
strategy=InterruptionStrategy.CONFIRMED,
min_speech_ms=300,
)
| Strategy | When to Use |
|---|
IMMEDIATE | Fast response, accept false positives |
CONFIRMED | Balanced — waits for sustained speech |
SEMANTIC | Ignore backchannel ("uh-huh") using BackchannelDetector |
DISABLED | Never interrupt (e.g., announcements) |
AudioFrame
Inbound audio is represented as AudioFrame:
from roomkit.voice.audio_frame import AudioFrame
frame = AudioFrame(
data=b"\x00" * 320, # Raw PCM bytes
sample_rate=16000, # Hz
channels=1, # Mono
sample_width=2, # 16-bit PCM
timestamp_ms=0.0,
)
Pipeline stages annotate frame.metadata as they process: denoiser, vad, aec, agc, diarization, dtmf keys.
Mock Providers for Testing
Every pipeline stage has a mock provider with pre-configured event sequences:
from roomkit.voice.pipeline import (
MockVADProvider,
MockDenoiserProvider,
MockDiarizationProvider,
MockAGCProvider,
MockAECProvider,
MockDTMFDetector,
MockAudioRecorder,
MockTurnDetector,
MockBackchannelDetector,
)
Example with mock VAD:
from roomkit.voice.pipeline import MockVADProvider, VADEvent, VADEventType
vad = MockVADProvider(events=[
VADEvent(type=VADEventType.SPEECH_START),
None,
VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech"),
])
RealtimeVoiceChannel connects to speech-to-speech AI models that handle audio directly — no separate STT/TTS needed. The AI model receives audio and responds with audio.
Basic Setup
from roomkit import RoomKit, ChannelCategory
from roomkit.channels.realtime_voice import RealtimeVoiceChannel
kit = RoomKit()
# Gemini Live example
from roomkit.voice import get_gemini_live_provider
GeminiLiveProvider = get_gemini_live_provider()
realtime = RealtimeVoiceChannel(
"realtime-voice",
provider=GeminiLiveProvider(
api_key="...",
model="gemini-2.0-flash-live-001",
),
system_prompt="You are a helpful voice assistant. Keep responses brief.",
)
kit.register_channel(realtime)
await kit.create_room(room_id="voice-room")
await kit.attach_channel("voice-room", "realtime-voice", category=ChannelCategory.INTELLIGENCE)
Providers
| Provider | Class | Extra | Description |
|---|
| Google Gemini Live | GeminiLiveProvider | roomkit[realtime-gemini] | Gemini 2.0 speech-to-speech |
| OpenAI Realtime | OpenAIRealtimeProvider | roomkit[realtime-openai] | GPT-4o realtime audio |
| xAI Grok | XAIRealtimeProvider | — | Grok speech-to-speech |
| Mock | MockRealtimeProvider | built-in | Testing |
# OpenAI Realtime
from roomkit.voice import get_openai_realtime_provider
OpenAIRealtime = get_openai_realtime_provider()
provider = OpenAIRealtime(api_key="sk-...", model="gpt-4o-realtime-preview")
# xAI Grok
from roomkit.voice import get_xai_realtime_provider, get_xai_realtime_config
XAIRealtime = get_xai_realtime_provider()
XAIConfig = get_xai_realtime_config()
provider = XAIRealtime(XAIConfig(api_key="..."))
Audio Transports
Transports handle the client-side audio (browser or device):
| Transport | Class | Extra | Use Case |
|---|
| WebSocket | WebSocketRealtimeTransport | roomkit[websocket] | Browser via WebSocket |
| FastRTC (WebRTC) | FastRTCRealtimeTransport | roomkit[fastrtc] | Browser via WebRTC |
| Local mic/speaker | LocalAudioBackend | roomkit[local-audio] | Local development |
| Mock | MockRealtimeTransport | built-in | Testing |
from roomkit.voice import get_websocket_realtime_transport
WSTransport = get_websocket_realtime_transport()
transport = WSTransport(host="0.0.0.0", port=8765)
realtime = RealtimeVoiceChannel(
"realtime-voice",
provider=provider,
transport=transport,
system_prompt="You are a helpful assistant.",
)
Joining a Session
session = await kit.join(
room_id="voice-room",
channel_id="realtime-voice",
participant_id="caller-1",
)
# Leave when done
await kit.leave(session)
Tool Calling
Realtime voice channels support tool calling during conversations:
from roomkit.channels.realtime_voice import RealtimeVoiceChannel, ToolHandler
async def handle_tool(name: str, arguments: dict) -> str:
if name == "get_weather":
return '{"temperature": 22, "condition": "sunny"}'
return '{"error": "unknown tool"}'
realtime = RealtimeVoiceChannel(
"realtime-voice",
provider=provider,
system_prompt="You help with weather. Use the get_weather tool.",
tools=[
{
"name": "get_weather",
"description": "Get weather for a city",
"parameters": {
"type": "object",
"properties": {"city": {"type": "string"}},
"required": ["city"],
},
},
],
tool_handler=handle_tool,
)
Text Injection
Inject text into an active realtime session (appears as if the AI said it):
session = await kit.join("room", "realtime-voice", participant_id="user")
await realtime.inject_text(session, "Let me check that for you...")
VAD Configuration
Configure voice activity detection for realtime providers:
realtime = RealtimeVoiceChannel(
"realtime-voice",
provider=provider,
system_prompt="...",
vad_config={
"threshold": 0.5,
"silence_duration_ms": 500,
},
)
Session Resumption
Some providers support session resumption after disconnection:
# OpenAI Realtime supports session resumption
provider = OpenAIRealtime(
api_key="sk-...",
model="gpt-4o-realtime-preview",
)
# Sessions automatically resume when possible
Hooks
from roomkit import HookTrigger, HookExecution
@kit.hook(HookTrigger.ON_TOOL_CALL, execution=HookExecution.ASYNC)
async def on_tool(event, ctx):
print(f"Realtime tool call: {event}")
@kit.hook(HookTrigger.ON_REALTIME_TEXT_INJECTED, execution=HookExecution.ASYNC)
async def on_inject(event, ctx):
print(f"Text injected: {event}")
RoomKit provides four declarative orchestration strategies for multi-agent workflows. Pass a strategy to RoomKit(orchestration=...) or create_room(orchestration=...) — agents, routing, handoff tools, and conversation state are wired automatically.
Strategies
Pipeline
Linear agent chain: triage -> handler -> resolver. Each agent can only hand off to the next in sequence.
from roomkit import Agent, Pipeline, RoomKit, WebSocketChannel
from roomkit.providers.ai.mock import MockAIProvider
triage = Agent(
"triage",
provider=MockAIProvider(responses=["Transferring you..."]),
role="Triage agent",
description="Routes requests to the right specialist",
system_prompt="You triage incoming requests.",
)
handler = Agent(
"handler",
provider=MockAIProvider(responses=["Let me help with that."]),
role="Request handler",
description="Handles customer requests",
system_prompt="You handle requests.",
)
resolver = Agent(
"resolver",
provider=MockAIProvider(responses=["All done!"]),
role="Resolution specialist",
description="Confirms resolution",
system_prompt="You resolve and close requests.",
)
kit = RoomKit(orchestration=Pipeline(agents=[triage, handler, resolver]))
Swarm
Every agent can hand off to every other agent. Bidirectional routing.
from roomkit import Swarm
kit = RoomKit(orchestration=Swarm(agents=[billing, shipping, returns]))
Supervisor
A supervisor agent delegates tasks to worker agents in child rooms:
from roomkit import Supervisor
kit = RoomKit(orchestration=Supervisor(
supervisor=manager_agent,
workers=[researcher, writer, reviewer],
))
Loop
Producer/reviewer cycle. The reviewer has an approve_output tool to break the loop.
from roomkit import Loop
kit = RoomKit(orchestration=Loop(
agent=writer_agent,
reviewer=editor_agent,
max_iterations=3,
))
Using Orchestration
from roomkit import InboundMessage, TextContent, WebSocketChannel
# Register transport channel
ws = WebSocketChannel("ws-user")
kit.register_channel(ws)
# Create room — orchestration auto-registers agents, creates router, sets initial state
await kit.create_room(room_id="support")
await kit.attach_channel("support", "ws-user")
# Messages are automatically routed to the active agent
await kit.process_inbound(
InboundMessage(
channel_id="ws-user",
sender_id="user",
content=TextContent(body="I need help with billing."),
)
)
Handoff Protocol
Agents hand off conversations by calling the handoff_conversation tool (auto-injected by orchestration strategies):
# The AI calls this tool automatically:
# handoff_conversation(target="handler", reason="Billing issue", summary="User needs invoice help")
The handoff:
- Updates
ConversationState in room metadata (phase, active agent, handoff count)
- Mutes the outgoing agent, unmutes the incoming agent
- Emits a system event visible to the new agent
- Records the transition in
phase_history
Conversation State
ConversationState tracks conversation progress within a room. It's stored in Room.metadata["_conversation_state"] and persists across all message turns. Orchestration strategies create and update it automatically, but you can also read and modify it directly.
ConversationState Fields
| Field | Type | Default | Description |
|---|
phase | str | "intake" | Current conversation phase. Can be any string — not restricted to built-in phases. |
active_agent_id | str | None | None | Channel ID of the currently active agent. |
previous_agent_id | str | None | None | Agent that was active before the last transition. |
handoff_count | int | 0 | Total number of agent handoffs. |
phase_started_at | datetime | now | Timestamp when the current phase started. |
phase_history | list[PhaseTransition] | [] | Immutable audit trail of all transitions. |
context | dict[str, Any] | {} | Arbitrary user data — store any custom key-value pairs across turns. |
Built-in phase constants (use any string — these are convenience defaults):
ConversationPhase.INTAKE, QUALIFICATION, HANDLING, ESCALATION, RESOLUTION, FOLLOWUP.
Reading State
from roomkit.orchestration.state import get_conversation_state
room = await kit.get_room("support")
state = get_conversation_state(room)
print(state.phase) # "handling"
print(state.active_agent_id) # "billing-agent"
print(state.handoff_count) # 2
print(state.context) # {"customer_tier": "premium", "issue_type": "refund"}
# Phase transition history (audit trail)
for t in state.phase_history:
print(f"{t.from_phase} -> {t.to_phase} by {t.from_agent} -> {t.to_agent} ({t.reason})")
Persisting Custom Data Across Turns
Use state.context to store arbitrary data that survives across conversation turns. Update state via set_conversation_state() + store.update_room():
from roomkit.orchestration.state import get_conversation_state, set_conversation_state
# Read current state
room = await kit.get_room("support")
state = get_conversation_state(room)
# Store custom data in context (immutable update pattern)
updated_state = state.model_copy(update={
"context": {
**state.context,
"customer_tier": "premium",
"issue_type": "refund",
"attempts": state.context.get("attempts", 0) + 1,
}
})
# Persist back to room
updated_room = set_conversation_state(room, updated_state)
await kit.store.update_room(updated_room)
Retrieving Custom Data on Later Turns
# In a hook or handler on a subsequent turn:
room = await kit.get_room("support")
state = get_conversation_state(room)
tier = state.context.get("customer_tier", "standard")
attempts = state.context.get("attempts", 0)
Programmatic Phase Transitions
Use state.transition() to change phase and record an audit entry. It returns a new immutable state:
from roomkit.orchestration.state import get_conversation_state, set_conversation_state
room = await kit.get_room("support")
state = get_conversation_state(room)
# Transition to a new phase with reason and agent
new_state = state.transition(
to_phase="escalation",
to_agent="supervisor-agent",
reason="Customer requested manager",
metadata={"escalation_priority": "high"},
)
# new_state.phase == "escalation"
# new_state.active_agent_id == "supervisor-agent"
# new_state.previous_agent_id == <old agent>
# new_state.handoff_count incremented if agent changed
# Persist
updated_room = set_conversation_state(room, new_state)
await kit.store.update_room(updated_room)
Using State in Hooks
A common pattern — persist data via a BEFORE_BROADCAST hook:
from roomkit import HookTrigger, HookResult, RoomEvent, RoomContext, TextContent
from roomkit.orchestration.state import get_conversation_state, set_conversation_state
@kit.hook(HookTrigger.BEFORE_BROADCAST)
async def track_sentiment(event: RoomEvent, ctx: RoomContext) -> HookResult:
if not isinstance(event.content, TextContent):
return HookResult.allow()
state = get_conversation_state(ctx.room)
# Update context with per-turn data
updated_state = state.model_copy(update={
"context": {
**state.context,
"last_message_length": len(event.content.body),
"message_count": state.context.get("message_count", 0) + 1,
}
})
updated_room = set_conversation_state(ctx.room, updated_state)
await kit.store.update_room(updated_room)
return HookResult.allow()
PhaseTransition Audit Record
Each transition creates an immutable PhaseTransition:
| Field | Type | Description |
|---|
from_phase | str | Previous phase |
to_phase | str | New phase |
from_agent | str | None | Previous agent channel ID |
to_agent | str | None | New agent channel ID |
reason | str | Why the transition occurred |
timestamp | datetime | When the transition occurred |
metadata | dict[str, Any] | Arbitrary metadata for this transition |
Conversation Router (Advanced)
ConversationRouter dynamically routes incoming messages to different agents based on conversation state, message content, origin channel, or custom logic. It works as a BEFORE_BROADCAST hook that stamps routing metadata on events.
RoutingConditions Reference
All conditions are ANDed — every non-None field must match for the rule to fire:
| Field | Type | Description |
|---|
phases | set[str] | None | Match when state.phase is in this set |
channel_types | set[ChannelType] | None | Match when sender's channel type is in this set |
intents | set[str] | None | Match when event.metadata["intent"] is in this set |
source_channel_ids | set[str] | None | Match when sender's channel ID is in this set |
custom | Callable | None | Custom function (event, context, state) -> bool for arbitrary logic |
Routing by Phase
from roomkit.orchestration.router import ConversationRouter, RoutingRule, RoutingConditions
router = ConversationRouter(
rules=[
RoutingRule(
agent_id="billing-agent",
conditions=RoutingConditions(phases={"billing"}),
priority=0,
),
RoutingRule(
agent_id="shipping-agent",
conditions=RoutingConditions(phases={"shipping"}),
priority=0,
),
],
default_agent_id="triage-agent",
)
Routing by Channel Type (Origin)
Route messages to different agents based on where they came from:
from roomkit.models.enums import ChannelType
router = ConversationRouter(
rules=[
RoutingRule(
agent_id="voice-specialist",
conditions=RoutingConditions(channel_types={ChannelType.VOICE}),
),
RoutingRule(
agent_id="sms-agent",
conditions=RoutingConditions(channel_types={ChannelType.SMS, ChannelType.WHATSAPP}),
),
],
default_agent_id="general-agent",
)
Routing by Intent (Content-Based)
Set event.metadata["intent"] via a classification hook, then route by intent:
@kit.hook(HookTrigger.BEFORE_BROADCAST, priority=-200) # Run before router
async def classify_intent(event: RoomEvent, ctx: RoomContext) -> HookResult:
if isinstance(event.content, TextContent):
body = event.content.body.lower()
intent = "billing" if "invoice" in body or "charge" in body else "general"
modified = event.model_copy(update={"metadata": {**(event.metadata or {}), "intent": intent}})
return HookResult.modify(modified)
return HookResult.allow()
router = ConversationRouter(
rules=[
RoutingRule(
agent_id="billing-agent",
conditions=RoutingConditions(intents={"billing"}),
),
],
default_agent_id="general-agent",
)
Routing with Custom Logic
Use a custom callable for complex routing decisions, including content-based or state-based logic:
def is_high_value_customer(event: RoomEvent, ctx: RoomContext, state: ConversationState) -> bool:
return state.context.get("customer_tier") == "premium"
def contains_urgency(event: RoomEvent, ctx: RoomContext, state: ConversationState) -> bool:
if isinstance(event.content, TextContent):
return any(word in event.content.body.lower() for word in ["urgent", "emergency", "asap"])
return False
router = ConversationRouter(
rules=[
# Premium customers with urgent messages -> senior agent
RoutingRule(
agent_id="senior-agent",
conditions=RoutingConditions(custom=is_high_value_customer),
priority=-1, # Check first
),
# Any urgent message -> escalation agent
RoutingRule(
agent_id="escalation-agent",
conditions=RoutingConditions(custom=contains_urgency),
priority=0,
),
],
default_agent_id="general-agent",
supervisor_id="supervisor-agent", # Always observes all conversations
)
Combined Conditions
Combine multiple conditions (all are ANDed):
# Only route SMS messages during billing phase to the billing specialist
RoutingRule(
agent_id="sms-billing-agent",
conditions=RoutingConditions(
phases={"billing"},
channel_types={ChannelType.SMS},
),
priority=0,
)
Installing the Router
# Option 1: Manual hook installation
kit.hook(
HookTrigger.BEFORE_BROADCAST,
execution=HookExecution.SYNC,
priority=-100,
)(router.as_hook())
# Option 2: Use install() which also sets up handoff tools
handler = router.install(kit, agents=[billing_agent, shipping_agent, triage_agent])
Routing Selection Priority
The router evaluates in this order:
- Agent affinity — if
state.active_agent_id is set and the agent is still attached, stick with it
- Rules — evaluate
RoutingRule list in ascending priority order; first match wins
- Fallback — return
default_agent_id
- Loop prevention — events FROM intelligence channels are never routed (returns
None)
Conversation Pipeline (Advanced)
Define stages explicitly:
from roomkit.orchestration.pipeline import ConversationPipeline, PipelineStage
pipeline = ConversationPipeline(stages=[
PipelineStage(phase="triage", agent_id="triage", next="handling"),
PipelineStage(phase="handling", agent_id="handler", next="resolution"),
PipelineStage(phase="resolution", agent_id="resolver"),
])
Status Bus
Agents can publish status updates for UI display:
# Agents publish status via the status bus
# Subscribe to updates
async def on_status(update):
print(f"Agent {update.agent_id}: {update.message} ({update.level})")
await kit.status_bus.subscribe(on_status)
Delegation
Delegate tasks to background agents in child rooms:
result = await kit.delegate(
room_id="main-room",
agent_id="researcher",
task="Find the latest pricing for product X",
context={"product": "X"},
)
Memory Providers
Agents use memory providers to maintain conversation context across handoffs:
from roomkit.memory.sliding_window import SlidingWindowMemory
from roomkit.orchestration.handoff import HandoffMemoryProvider
agent = Agent(
"agent",
provider=provider,
memory=HandoffMemoryProvider(SlidingWindowMemory(max_events=50)),
)
HandoffMemoryProvider wraps any memory provider to inject handoff context (summary from the previous agent) into the conversation history.
Transport providers handle sending and receiving messages over external protocols. Each provider implements a channel-specific ABC.
SMS
Twilio
from roomkit import RoomKit, SMSChannel
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig
sms = SMSChannel("sms-twilio", provider=TwilioSMSProvider(TwilioConfig(
account_sid="AC...",
auth_token="...",
from_number="+15551234567",
)))
kit = RoomKit()
kit.register_channel(sms)
Telnyx
from roomkit.providers.telnyx.sms import TelnyxSMSProvider
from roomkit.providers.telnyx.config import TelnyxConfig
sms = SMSChannel("sms-telnyx", provider=TelnyxSMSProvider(TelnyxConfig(
api_key="KEY...",
from_number="+15551234567",
)))
Sinch
from roomkit.providers.sinch.sms import SinchSMSProvider
from roomkit.providers.sinch.config import SinchConfig
sms = SMSChannel("sms-sinch", provider=SinchSMSProvider(SinchConfig(
service_plan_id="...",
api_token="...",
from_number="+15551234567",
)))
VoiceMeUp
from roomkit.providers.voicemeup.sms import VoiceMeUpSMSProvider
from roomkit.providers.voicemeup.config import VoiceMeUpConfig
sms = SMSChannel("sms-vmu", provider=VoiceMeUpSMSProvider(VoiceMeUpConfig(
username="...",
auth_token="...",
from_number="+15551234567",
)))
Webhook Parsing
Each SMS provider has a webhook parser:
from roomkit.providers.twilio.sms import parse_twilio_webhook
from roomkit.providers.telnyx.sms import parse_telnyx_webhook
from roomkit.providers.sinch.sms import parse_sinch_webhook
from roomkit.providers.voicemeup.sms import parse_voicemeup_webhook
# Or use the universal webhook parser
message = await kit.process_webhook(meta=request_data, channel_id="sms-twilio")
RCS
from roomkit import RCSChannel
from roomkit.providers.twilio.rcs import TwilioRCSProvider, TwilioRCSConfig
rcs = RCSChannel("rcs-main", provider=TwilioRCSProvider(TwilioRCSConfig(
account_sid="AC...",
auth_token="...",
messaging_service_sid="MG...", # Required for RCS (must be RCS-enabled)
)))
Also available via Telnyx: TelnyxRCSProvider, TelnyxRCSConfig.
Email
Elastic Email
from roomkit import EmailChannel
from roomkit.providers.elasticemail.email import ElasticEmailProvider
from roomkit.providers.elasticemail.config import ElasticEmailConfig
email = EmailChannel("email-main", provider=ElasticEmailProvider(ElasticEmailConfig(
api_key="...",
from_email="support@example.com",
from_name="Support Team",
)))
SendGrid
from roomkit.providers.sendgrid.email import SendGridEmailProvider
from roomkit.providers.sendgrid.config import SendGridConfig
email = EmailChannel("email-sg", provider=SendGridEmailProvider(SendGridConfig(
api_key="SG...",
from_email="support@example.com",
)))
WhatsApp
Business API (Cloud)
from roomkit import WhatsAppChannel
from roomkit.providers.whatsapp.base import WhatsAppProvider
whatsapp = WhatsAppChannel("wa-business", provider=WhatsAppProvider(
access_token="...",
phone_number_id="...",
))
Personal (neonize)
from roomkit import WhatsAppPersonalChannel
from roomkit.providers.whatsapp.personal import WhatsAppPersonalProvider
whatsapp = WhatsAppPersonalChannel("wa-personal", provider=WhatsAppPersonalProvider())
Requires pip install roomkit[whatsapp-personal]. Uses the neonize library for multidevice protocol with typing indicators, read receipts, and media handling.
Facebook Messenger
from roomkit import MessengerChannel
from roomkit.providers.messenger.facebook import FacebookMessengerProvider
from roomkit.providers.messenger.config import MessengerConfig
messenger = MessengerChannel("messenger", provider=FacebookMessengerProvider(MessengerConfig(
page_access_token="...",
app_secret="...",
verify_token="...",
)))
Webhook parser: parse_messenger_webhook(request_data, channel_id="messenger") — returns list[InboundMessage].
Telegram
from roomkit import TelegramChannel
from roomkit.providers.telegram.bot import TelegramBotProvider
from roomkit.providers.telegram.config import TelegramConfig
telegram = TelegramChannel("telegram", provider=TelegramBotProvider(TelegramConfig(
bot_token="123456:ABC-DEF...",
)))
Webhook parser: parse_telegram_webhook(request_data, channel_id="telegram") — returns list[InboundMessage].
Microsoft Teams
from roomkit import TeamsChannel
from roomkit.providers.teams.bot_framework import BotFrameworkTeamsProvider
from roomkit.providers.teams.config import TeamsConfig
teams = TeamsChannel("teams", provider=BotFrameworkTeamsProvider(TeamsConfig(
app_id="...",
app_password="...",
)))
Features: proactive messaging, bot mention detection, reaction handling, conversation reference storage.
from roomkit.providers.teams.webhook import parse_teams_webhook, is_bot_added
# Parse incoming Teams activity
activity = parse_teams_webhook(request_data)
# Check if bot was added to a conversation
if is_bot_added(activity):
# Handle bot installation
pass
HTTP (Generic Webhook)
from roomkit import HTTPChannel
from roomkit.providers.http.provider import WebhookHTTPProvider
from roomkit.providers.http.config import HTTPProviderConfig
http = HTTPChannel("webhook", provider=WebhookHTTPProvider(HTTPProviderConfig(
url="https://api.example.com/messages",
headers={"Authorization": "Bearer ..."},
)))
WebSocket
WebSocket channels don't use a provider — they handle connections directly:
from roomkit import WebSocketChannel
ws = WebSocketChannel("ws-client")
# Register a connection
ws.register_connection("conn-1", on_receive_callback)
# In production, connect to the framework
await kit.connect_websocket("ws-client", "conn-1", send_fn)
await kit.disconnect_websocket("ws-client", "conn-1")
Phone Number Utilities
from roomkit.providers.sms.phone import is_valid_phone, normalize_phone
is_valid_phone("+15551234567") # True
normalize_phone("555-123-4567") # "+15551234567"
Delivery Status Tracking
Track delivery status for sent messages:
from roomkit import DeliveryStatus
@kit.on_delivery_status
async def track_delivery(status: DeliveryStatus) -> None:
if status.status == "failed":
print(f"Message {status.message_id} failed: {status.error_message}")
# Process status webhooks from providers
await kit.process_delivery_status(status)
Identity Resolution
The identity pipeline maps external sender IDs to known participants. It runs as part of the inbound pipeline, after handle_inbound() and before hooks.
How It Works
Inbound message arrives with sender_id
-> IdentityResolver.resolve(sender_id, channel_type)
-> Returns IdentityResult with status:
IDENTIFIED -> participant_id stamped on event, processing continues
AMBIGUOUS -> ON_IDENTITY_AMBIGUOUS hook fires
PENDING -> ON_IDENTITY_AMBIGUOUS hook fires
UNKNOWN -> ON_IDENTITY_UNKNOWN hook fires
REJECTED -> ON_IDENTITY_UNKNOWN hook fires
Identity Hooks
from roomkit import RoomKit, HookTrigger
from roomkit.models.identity import IdentityHookResult, Identity
kit = RoomKit()
@kit.identity_hook(HookTrigger.ON_IDENTITY_UNKNOWN)
async def handle_unknown(event, ctx):
# Option 1: Resolve to a known identity
return IdentityHookResult.resolved(Identity(
id="user-123",
display_name="Alice",
))
# Option 2: Challenge the sender to identify
return IdentityHookResult.challenge(inject=InjectedEvent(
content=TextContent(body="Please provide your account number."),
))
# Option 3: Reject the message
return IdentityHookResult.reject("Unknown sender")
# Option 4: Keep as pending
return IdentityHookResult.pending(candidates=[...])
Custom Identity Resolver
from roomkit.identity.base import IdentityResolver
from roomkit.models.identity import IdentityResult, Identity
from roomkit.models.enums import IdentificationStatus
class DatabaseIdentityResolver(IdentityResolver):
async def resolve(self, message: InboundMessage, context: RoomContext) -> IdentityResult:
user = await db.find_by_phone(message.sender_id)
if user:
return IdentityResult(
status=IdentificationStatus.IDENTIFIED,
identity=Identity(id=user.id, display_name=user.name),
)
return IdentityResult(status=IdentificationStatus.UNKNOWN)
kit = RoomKit(identity_resolver=DatabaseIdentityResolver())
Manual Resolution
# Resolve a pending participant to a known identity
await kit.resolve_participant(
room_id="room-1",
participant_id="pending-123",
identity_id="user-456",
)
Realtime Ephemeral Events
Ephemeral events (typing, presence, reactions) are not stored in conversation history. They're delivered in real-time to subscribers.
Publishing Events
from roomkit import RoomKit
kit = RoomKit()
# Typing indicator
await kit.publish_typing("room-1", "alice", is_typing=True)
await kit.publish_typing("room-1", "alice", is_typing=False)
# Presence
await kit.publish_presence("room-1", "alice", "online") # online/away/offline
# Reaction
await kit.publish_reaction("room-1", "alice", target_event_id="evt-123", emoji="thumbsup")
# Read receipt
await kit.publish_read_receipt("room-1", "alice", event_id="evt-123")
# Tool call events (AIChannel publishes these automatically)
from roomkit.realtime.base import EphemeralEventType
await kit.publish_tool_call("room-1", "ai-agent", [
{"id": "tc1", "name": "search", "arguments": {"q": "test"}}
], EphemeralEventType.TOOL_CALL_START)
Subscribing to Events
async def on_ephemeral(event):
print(f"Ephemeral: {event.type} from {event.user_id}")
sub_id = await kit.subscribe_room("room-1", on_ephemeral)
# Unsubscribe later
await kit.unsubscribe_room(sub_id)
Event Types
| Type | Description |
|---|
TYPING_START | User started typing |
TYPING_STOP | User stopped typing |
PRESENCE_ONLINE | User came online |
PRESENCE_AWAY | User went away |
PRESENCE_OFFLINE | User went offline |
READ_RECEIPT | User read a message |
REACTION | User reacted to a message |
TOOL_CALL_START | AI started executing a tool |
TOOL_CALL_END | AI finished executing a tool |
CUSTOM | Custom ephemeral event |
Read Tracking
# Mark a specific event as read
await kit.mark_read("room-1", "ws-user", "evt-123")
# Mark all events as read
await kit.mark_all_read("room-1", "ws-user")
Custom Realtime Backend
The default InMemoryRealtime works for single-process deployments. For distributed systems, implement RealtimeBackend:
from roomkit.realtime.base import RealtimeBackend
class RedisRealtimeBackend(RealtimeBackend):
# Implement publish, subscribe, unsubscribe using Redis Pub/Sub
...
kit = RoomKit(realtime=RedisRealtimeBackend())
RoomKit provides built-in resilience patterns for production deployments: rate limiting, circuit breakers, retry with backoff, chain depth limits, and room lifecycle timers.
Rate Limiting
Apply rate limits per channel binding:
from roomkit import RoomKit
from roomkit.models.channel import RateLimit
kit = RoomKit()
await kit.attach_channel("room-1", "sms-out",
rate_limit=RateLimit(max_per_second=1.0, max_per_minute=30.0),
)
Framework-level inbound rate limiting:
from roomkit import RoomKit
from roomkit.models.channel import RateLimit
kit = RoomKit(inbound_rate_limit=RateLimit(max_per_second=10.0))
Circuit Breaker
Isolate provider failures with circuit breakers:
from roomkit.core.circuit_breaker import CircuitBreaker
cb = CircuitBreaker(failure_threshold=5, recovery_timeout=60.0)
if cb.allow_request():
try:
result = await provider.send(event, to="+1234567890")
cb.record_success()
except Exception:
cb.record_failure() # Opens after 5 consecutive failures
States: CLOSED (normal) -> OPEN (failing, fast-reject) -> HALF_OPEN (testing recovery).
Retry with Backoff
Retry failed operations with exponential backoff:
from roomkit.models.channel import RetryPolicy
from roomkit.core.retry import retry_with_backoff
policy = RetryPolicy(
max_retries=3,
base_delay_seconds=1.0,
max_delay_seconds=60.0,
)
result = await retry_with_backoff(flaky_function, policy)
Apply per binding:
await kit.attach_channel("room-1", "sms-out",
retry_policy=RetryPolicy(max_retries=3, base_delay_seconds=1.0),
)
Chain Depth Limit
Prevents infinite AI-to-AI loops. Default max depth is 5.
kit = RoomKit(max_chain_depth=3) # Stricter limit
When an AI response triggers another AI response, chain depth increments. Processing stops when the limit is reached.
Room Lifecycle Timers
Auto-transition rooms based on inactivity:
from roomkit import RoomKit
kit = RoomKit()
room = await kit.create_room(room_id="session-1")
# Configure timers on the room:
# inactive_after_seconds -> ACTIVE to PAUSED
# closed_after_seconds -> to CLOSED
# Check timers for one room
room = await kit.check_room_timers("session-1")
# Batch check all rooms (call periodically)
transitioned = await kit.check_all_timers()
Delivery Status Tracking
Track whether messages were delivered to providers:
from roomkit import DeliveryStatus
@kit.on_delivery_status
async def track(status: DeliveryStatus) -> None:
if status.status == "failed":
logger.error("Delivery failed: %s — %s", status.message_id, status.error_message)
elif status.status == "delivered":
logger.info("Delivered: %s", status.message_id)
# Process status webhooks from providers
await kit.process_delivery_status(status)
Production Setup Example
from roomkit import RoomKit
from roomkit.models.channel import RateLimit, RetryPolicy
from roomkit.store.postgres import PostgresStore
kit = RoomKit(
store=PostgresStore("postgresql://user:pass@localhost/roomkit"),
max_chain_depth=5,
inbound_rate_limit=RateLimit(max_per_second=50.0),
process_timeout=30.0,
)
# Per-channel resilience
await kit.attach_channel("room", "sms-out",
rate_limit=RateLimit(max_per_second=1.0, max_per_minute=30.0),
retry_policy=RetryPolicy(max_retries=3, base_delay_seconds=1.0),
)
Framework Events for Monitoring
@kit.on("source_error")
async def on_error(event):
logger.error("Source error: %s", event.data["error"])
@kit.on("voice_session_ended")
async def on_voice_end(event):
logger.info("Voice session ended: %s", event.data["session_id"])
RoomKit uses the ConversationStore ABC for persistence. The default InMemoryStore works out of the box. For production, use PostgresStore.
InMemoryStore (Default)
from roomkit import RoomKit
kit = RoomKit() # Uses InMemoryStore automatically
Data lives in Python dicts — fast for development, lost on restart.
PostgresStore
Install: pip install roomkit[postgres]
from roomkit import RoomKit
from roomkit.store.postgres import PostgresStore
store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init() # Creates tables if they don't exist
kit = RoomKit(store=store)
Connection Pooling
store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init(min_size=5, max_size=20) # Pool sizing via init()
Schema
PostgresStore creates 10 tables:
| Table | Purpose |
|---|
rooms | Room records with status, metadata, timers |
events | Event timeline with sequential indexing |
participants | Room participants with roles and status |
bindings | Channel-to-room bindings with config |
identities | Known identity records |
tasks | AI-extracted tasks |
observations | AI-extracted observations |
delivery_status | Message delivery tracking |
read_tracking | Per-channel read positions |
telemetry_spans | Telemetry span records |
Operations
# Room operations
room = await kit.create_room(room_id="persistent", metadata={"topic": "billing"})
# Event storage and retrieval
events = await kit.store.list_events("persistent", offset=0, limit=50)
# Timeline query with filters
timeline = await kit.get_timeline("persistent", offset=0, limit=50)
# Participant management
participants = await kit.store.list_participants("persistent")
# Binding management
bindings = await kit.store.list_bindings("persistent")
Full Example
from __future__ import annotations
import asyncio
import os
from roomkit import InboundMessage, RoomKit, TextContent, WebSocketChannel
from roomkit.store.postgres import PostgresStore
async def main() -> None:
store = PostgresStore(os.environ["DATABASE_URL"])
await store.init()
kit = RoomKit(store=store)
ws = WebSocketChannel("ws-user")
kit.register_channel(ws)
room = await kit.create_room(room_id="support", metadata={"topic": "billing"})
await kit.attach_channel("support", "ws-user")
await kit.process_inbound(
InboundMessage(
channel_id="ws-user",
sender_id="user",
content=TextContent(body="I need help with my invoice"),
)
)
# Data persists across restarts
events = await kit.store.list_events("support")
print(f"Stored {len(events)} events")
await kit.close()
if __name__ == "__main__":
asyncio.run(main())
Custom Store
Implement ConversationStore for other backends (Redis, DynamoDB, etc.):
from roomkit.store.base import ConversationStore
from roomkit.models.room import Room
class RedisStore(ConversationStore):
async def create_room(self, room: Room) -> Room:
await self.redis.set(f"room:{room.id}", room.model_dump_json())
return room
# Implement all abstract methods...
kit = RoomKit(store=RedisStore())
RoomKit provides mock implementations for every pluggable component: AI providers, voice backends, pipeline stages, identity resolvers, and telemetry.
Test Setup
pip install roomkit[dev]
uv run pytest # Run all tests
uv run pytest tests/test_framework.py -v # Specific file
Configuration in pyproject.toml:
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto" # No @pytest.mark.asyncio needed
Basic Test Pattern
from roomkit import RoomKit, InboundMessage, TextContent, WebSocketChannel
class TestMyFeature:
async def test_message_delivery(self) -> None:
kit = RoomKit()
ws = WebSocketChannel("ws-user")
kit.register_channel(ws)
inbox: list = []
async def on_recv(_conn: str, event) -> None:
inbox.append(event)
ws.register_connection("conn", on_recv)
await kit.create_room(room_id="test-room")
await kit.attach_channel("test-room", "ws-user")
result = await kit.process_inbound(
InboundMessage(
channel_id="ws-user",
sender_id="user",
content=TextContent(body="Hello"),
)
)
assert not result.blocked
assert len(inbox) == 1
assert inbox[0].content.body == "Hello"
Mock AI Provider
from roomkit.providers.ai.mock import MockAIProvider
from roomkit.channels.ai import AIChannel
from roomkit import ChannelCategory
# Responds with pre-configured messages in order
provider = MockAIProvider(responses=["Response 1", "Response 2"])
ai = AIChannel("ai", provider=provider)
kit.register_channel(ai)
await kit.attach_channel("room", "ai", category=ChannelCategory.INTELLIGENCE)
# After processing, check what the AI was called with:
assert len(provider.calls) == 1
last_call = provider.calls[-1]
print(last_call.system_prompt)
print(last_call.temperature)
print([t.name for t in last_call.tools])
Mock Voice Backend
from roomkit.voice.backends.mock import MockVoiceBackend
backend = MockVoiceBackend()
# Simulate audio input
backend.simulate_audio_received(session, audio_frame)
Mock Pipeline Providers
Every pipeline stage has a mock that accepts pre-configured event sequences:
from roomkit.voice.pipeline import (
MockVADProvider,
VADEvent,
VADEventType,
MockDenoiserProvider,
MockDiarizationProvider,
MockAGCProvider,
MockAECProvider,
MockDTMFDetector,
MockAudioRecorder,
MockTurnDetector,
MockBackchannelDetector,
)
# VAD with event sequence
vad = MockVADProvider(events=[
VADEvent(type=VADEventType.SPEECH_START),
None, # No event for this frame
VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"speech"),
])
# Other mocks
denoiser = MockDenoiserProvider()
diarizer = MockDiarizationProvider()
agc = MockAGCProvider()
aec = MockAECProvider()
dtmf = MockDTMFDetector()
recorder = MockAudioRecorder()
turn = MockTurnDetector()
backchannel = MockBackchannelDetector()
Mock STT/TTS
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
stt = MockSTTProvider(transcriptions=["Hello", "How are you?"])
tts = MockTTSProvider()
Mock Identity Resolver
from roomkit.identity.mock import MockIdentityResolver
resolver = MockIdentityResolver()
kit = RoomKit(identity_resolver=resolver)
Mock Realtime Provider
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport
provider = MockRealtimeProvider()
transport = MockRealtimeTransport()
Testing Hooks
from roomkit import HookTrigger, HookResult, HookExecution
async def test_hook_blocks_message() -> None:
kit = RoomKit()
ws = WebSocketChannel("ws")
kit.register_channel(ws)
await kit.create_room(room_id="r")
await kit.attach_channel("r", "ws")
@kit.hook(HookTrigger.BEFORE_BROADCAST)
async def blocker(event, ctx):
return HookResult.block("blocked")
result = await kit.process_inbound(
InboundMessage(
channel_id="ws",
sender_id="user",
content=TextContent(body="test"),
)
)
assert result.blocked
assert result.reason == "blocked"
Testing Voice Pipeline
from roomkit import VoiceChannel
from roomkit.voice.pipeline import AudioPipelineConfig, MockVADProvider, VADEvent, VADEventType
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.backends.mock import MockVoiceBackend
from roomkit.voice.audio_frame import AudioFrame
async def test_voice_pipeline() -> None:
kit = RoomKit()
backend = MockVoiceBackend()
stt = MockSTTProvider(transcriptions=["Hello"])
tts = MockTTSProvider()
vad = MockVADProvider(events=[
VADEvent(type=VADEventType.SPEECH_START),
None,
VADEvent(type=VADEventType.SPEECH_END, audio_bytes=b"audio"),
])
voice = VoiceChannel(
"voice", stt=stt, tts=tts, backend=backend,
pipeline=AudioPipelineConfig(vad=vad),
)
kit.register_channel(voice)
await kit.create_room(room_id="call")
await kit.attach_channel("call", "voice")
# Simulate audio input
frame = AudioFrame(data=b"\x00" * 320, sample_rate=16000)
backend.simulate_audio_received(None, frame)
Testing Orchestration
from roomkit import Agent, Pipeline, RoomKit, WebSocketChannel
from roomkit.providers.ai.mock import MockAIProvider
async def test_pipeline_orchestration() -> None:
agent1 = Agent("a1", provider=MockAIProvider(responses=["Transferring..."]))
agent2 = Agent("a2", provider=MockAIProvider(responses=["Resolved!"]))
kit = RoomKit(orchestration=Pipeline(agents=[agent1, agent2]))
ws = WebSocketChannel("ws")
kit.register_channel(ws)
await kit.create_room(room_id="test")
await kit.attach_channel("test", "ws")
# First message goes to agent1
result = await kit.process_inbound(
InboundMessage(channel_id="ws", sender_id="user", content=TextContent(body="Help"))
)
assert not result.blocked
Test Utilities
# Pydantic model updates (immutable)
modified = event.model_copy(update={"content": TextContent(body="new")})
# Never mutate models directly — always use model_copy
RoomKit exports 73 symbols from roomkit. Providers and voice types import from subpackages.
Top-Level Imports (from roomkit import ...)
Framework
| Symbol | Description |
|---|
RoomKit | Central orchestrator — rooms, channels, hooks, storage |
Channels
| Symbol | Description |
|---|
Agent | AI agent with role, description, greeting, tools |
AIChannel | Intelligence layer for AI responses |
AudioVideoChannel | Combined audio + video channel |
Channel | Base class for all channels |
CLIChannel | CLI/terminal interactive channel |
EmailChannel | Email transport channel factory |
HTTPChannel | HTTP webhook transport channel factory |
MessengerChannel | Facebook Messenger transport channel factory |
RCSChannel | RCS transport channel factory |
RealtimeAudioVideoChannel | Realtime speech-to-speech with video |
RealtimeVoiceChannel | Speech-to-speech AI channel |
SMSChannel | SMS transport channel factory |
TeamsChannel | Microsoft Teams transport channel factory |
TelegramChannel | Telegram Bot transport channel factory |
TransportChannel | Generic transport channel wrapper |
VideoChannel | Video channel with vision pipeline |
VoiceChannel | Real-time audio with STT/TTS/pipeline |
WebSocketChannel | WebSocket bidirectional channel |
WhatsAppChannel | WhatsApp Business API channel factory |
WhatsAppPersonalChannel | WhatsApp Personal (neonize) channel factory |
Enums
| Symbol | Description |
|---|
Access | Channel access levels: READ_WRITE, READ_ONLY, WRITE_ONLY, NONE |
ChannelCategory | TRANSPORT or INTELLIGENCE |
ChannelType | SMS, EMAIL, WHATSAPP, VOICE, AI, WEBSOCKET, etc. |
EventStatus | DELIVERED, BLOCKED, etc. |
EventType | MESSAGE, SYSTEM, EDIT, DELETE, etc. |
HookExecution | SYNC or ASYNC |
HookTrigger | 40+ hook triggers (BEFORE_BROADCAST, AFTER_BROADCAST, etc.) |
RoomStatus | ACTIVE, PAUSED, CLOSED, ARCHIVED |
Orchestration
| Symbol | Description |
|---|
Loop | Producer/reviewer cycle strategy |
Orchestration | ABC for orchestration strategies |
Pipeline | Linear agent chain strategy |
Supervisor | Supervisor delegates to workers strategy |
Swarm | Bidirectional handoff strategy |
Models
| Symbol | Description |
|---|
ChannelBinding | Binding of a channel to a room |
ChannelCapabilities | Declared capabilities of a channel |
ChannelOutput | Output of a channel delivery |
DeliveryResult | Result of delivering a message |
DeliveryStatus | Delivery status from provider webhook |
EventSource | Source attribution for an event |
FrameworkEvent | Lightweight framework lifecycle event |
HookResult | Result from sync hooks: .allow(), .block(reason), .modify(event) |
InjectedEvent | Event injected by a hook |
InboundMessage | Incoming message from a provider |
InboundResult | Result of processing an inbound message |
Participant | Participant data model |
ProviderResult | Result from a provider operation |
Room | Room data model |
RoomContext | Context passed to hooks (room, bindings, participants, events) |
RoomEvent | Core event stored in the timeline |
RoomTimers | Timer configuration for room inactivity |
SessionStartedEvent | Event fired when a voice session starts |
TextContent | Plain text content |
Tool | Base class for tool definitions |
ToolCallCallback | Callback type for tool call events |
ToolCallEvent | Tool call event model |
ToolHandler | Tool handler type for realtime voice |
get_current_voice_session | Get the current voice session from context |
Errors
| Symbol | Description |
|---|
RoomKitError | Base exception |
RoomNotFoundError | Room does not exist |
ChannelNotFoundError | Channel not attached to room |
ChannelNotRegisteredError | Channel not registered with framework |
ParticipantNotFoundError | Participant not found in room |
IdentityNotFoundError | Identity not found |
SourceAlreadyAttachedError | Source already attached |
SourceNotFoundError | No source attached |
VoiceBackendNotConfiguredError | Voice backend not configured |
VoiceNotConfiguredError | Voice (STT/TTS) not configured |
AI Documentation Helpers
| Symbol | Description |
|---|
get_llms_txt() | Get llms.txt content |
get_llms_full_txt() | Get llms-full.txt content (comprehensive) |
get_agents_md() | Get AGENTS.md content |
get_ai_context() | Get combined AI context |
RoomKit Constructor
kit = RoomKit(
store=InMemoryStore(), # ConversationStore implementation
identity_resolver=None, # IdentityResolver implementation
identity_channel_types=None, # Channel types to resolve identity for
inbound_router=None, # InboundRoomRouter implementation
lock_manager=None, # RoomLockManager implementation
realtime=None, # RealtimeBackend implementation
max_chain_depth=5, # AI-to-AI loop prevention
identity_timeout=10.0, # Identity resolution timeout (seconds)
process_timeout=30.0, # Inbound processing timeout (seconds)
stt=None, # STTProvider for transcription
tts=None, # TTSProvider for synthesis
voice=None, # VoiceBackend for audio transport
task_runner=None, # Background task runner
delivery_strategy=None, # Delivery strategy
status_bus=None, # Status bus for orchestration
telemetry=None, # TelemetryProvider
inbound_rate_limit=None, # Framework-level rate limit
orchestration=None, # Orchestration strategy
)
Key RoomKit Methods
Room Lifecycle
| Method | Description |
|---|
create_room(room_id?, metadata?, orchestration?) | Create a room |
get_room(room_id) | Get room by ID |
close_room(room_id) | Close a room |
update_room_metadata(room_id, metadata) | Update room metadata |
check_room_timers(room_id) | Check timer transitions for one room |
check_all_timers() | Check all room timers |
Channel Operations
| Method | Description |
|---|
register_channel(channel) | Register a channel |
attach_channel(room_id, channel_id, category?, access?, ...) | Attach channel to room |
detach_channel(room_id, channel_id) | Detach channel from room |
mute(room_id, channel_id) | Mute a channel |
unmute(room_id, channel_id) | Unmute a channel |
set_access(room_id, channel_id, access) | Set channel access level |
Voice/Video
| Method | Description |
|---|
join(room_id, channel_id, participant_id?, ...) | Join voice/video session |
leave(session) | Leave voice/video session |
transcribe(audio) | Speech-to-text |
synthesize(text, voice?) | Text-to-speech |
Inbound Pipeline
| Method | Description |
|---|
process_inbound(message, room_id?) | Process an inbound message |
Hooks
| Method | Description |
|---|
hook(trigger, execution?, priority?, ...) | Decorator to register a hook |
on(event_type) | Decorator for framework events |
identity_hook(trigger, ...) | Decorator for identity hooks |
on_delivery_status(fn) | Decorator for delivery status |
add_room_hook(room_id, trigger, execution, fn, ...) | Add room-scoped hook |
remove_room_hook(room_id, name) | Remove room-scoped hook |
Realtime
| Method | Description |
|---|
publish_typing(room_id, user_id, is_typing?) | Typing indicator |
publish_presence(room_id, user_id, status) | Presence update |
publish_reaction(room_id, user_id, target_event_id, emoji) | Reaction |
publish_read_receipt(room_id, user_id, event_id) | Read receipt |
subscribe_room(room_id, callback) | Subscribe to ephemeral events |
unsubscribe_room(subscription_id) | Unsubscribe |
Sources
| Method | Description |
|---|
attach_source(channel_id, source, auto_restart?, ...) | Attach event source |
detach_source(channel_id) | Detach event source |
source_health(channel_id) | Get source health |
Other
| Method | Description |
|---|
delegate(room_id, agent_id, task, ...) | Delegate to background agent |
send_greeting(room_id, channel_id?, greeting?, ...) | Send greeting |
send_event(room_id, channel_id, content, ...) | Send event directly |
get_timeline(room_id, offset?, limit?) | Query event timeline |
close() | Shutdown framework |
Provider Subpackage Imports
AI Providers
from roomkit.providers.anthropic.ai import AnthropicAIProvider
from roomkit.providers.anthropic.config import AnthropicConfig
from roomkit.providers.openai.ai import OpenAIAIProvider
from roomkit.providers.openai.config import OpenAIConfig
from roomkit.providers.gemini.ai import GeminiAIProvider
from roomkit.providers.gemini.config import GeminiConfig
from roomkit.providers.mistral.ai import MistralAIProvider
from roomkit.providers.mistral.config import MistralConfig
from roomkit.providers.ai.mock import MockAIProvider
from roomkit.providers.ai.base import AIProvider, AIContext, AIResponse, AITool, AIToolCall
SMS Providers
from roomkit.providers.twilio.sms import TwilioSMSProvider
from roomkit.providers.twilio.config import TwilioConfig
from roomkit.providers.telnyx.sms import TelnyxSMSProvider
from roomkit.providers.telnyx.config import TelnyxConfig
from roomkit.providers.sinch.sms import SinchSMSProvider
from roomkit.providers.sinch.config import SinchConfig
from roomkit.providers.sms.mock import MockSMSProvider
Voice (Lazy Loaders)
from roomkit.voice import (
get_deepgram_provider, get_deepgram_config,
get_elevenlabs_provider, get_elevenlabs_config,
get_sherpa_onnx_stt_provider, get_sherpa_onnx_tts_provider,
get_local_audio_backend,
get_fastrtc_backend,
get_rtp_backend,
get_sip_backend,
get_gemini_live_provider,
get_openai_realtime_provider,
get_xai_realtime_provider,
get_websocket_realtime_transport,
get_speex_aec_provider,
get_rnnoise_denoiser_provider,
)
Voice Mocks
from roomkit.voice.backends.mock import MockVoiceBackend
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport
Pipeline
from roomkit.voice.pipeline import (
AudioPipelineConfig, VADConfig,
MockVADProvider, VADEvent, VADEventType,
MockDenoiserProvider, MockDiarizationProvider,
MockAGCProvider, MockAECProvider, MockDTMFDetector,
MockAudioRecorder, MockTurnDetector, MockBackchannelDetector,
)
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
from roomkit.voice.audio_frame import AudioFrame
Orchestration
from roomkit.orchestration.state import get_conversation_state, ConversationState
from roomkit.orchestration.router import ConversationRouter, RoutingRule
from roomkit.orchestration.pipeline import ConversationPipeline, PipelineStage
from roomkit.orchestration.handoff import HandoffHandler, HandoffMemoryProvider
Storage
from roomkit.store.base import ConversationStore
from roomkit.store.memory import InMemoryStore
from roomkit.store.postgres import PostgresStore
Content Types
from roomkit.models.event import (
TextContent, RichContent, MediaContent, AudioContent, VideoContent,
LocationContent, CompositeContent, TemplateContent, SystemContent,
EditContent, DeleteContent,
)
