# RoomKit
> 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.
> Python 3.12+, Pydantic 2.x, fully typed, zero required dependencies beyond Pydantic.
> Minimum version: 0.66.1 | Python 3.12+
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, WhatsApp, Telegram, Discord, Buzz/Nostr, Voice, Video, Conference/SFU, AI, ACP coding agents, WebSocket, CLI, etc. — 23 `ChannelType` values). Registered once, attached to many rooms.
- **Hook** — A function that intercepts events at specific points in the pipeline (76 triggers). 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
- **AI conference participants** — a bot that joins a multi-party SFU video conference (LiveKit) with STT/TTS, transcription, and recording
- **Coding-agent frontends** — drive Claude Code or any Agent Client Protocol agent from SMS, Telegram, or the terminal via `ACPChannel`
- **Nostr workspace bots** — chat and huddle voice over Buzz relay channels
## 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** — CI-enforced `ty` checks, full type hints, Pydantic models throughout.
- **Python 3.12+** — Uses modern syntax (`X | None`, not `Optional[X]`).
---
## Basic Install
```bash
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
```bash
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
```bash
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
```bash
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
```bash
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
```bash
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)
```bash
pip install roomkit[realtime-openai] # OpenAI Realtime API
pip install roomkit[realtime-gemini] # Google Gemini Live API
```
### Messaging Providers
```bash
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
```bash
pip install roomkit[postgres] # PostgreSQL persistence (asyncpg)
pip install roomkit[mcp] # Model Context Protocol tools
pip install roomkit[opentelemetry] # OpenTelemetry tracing
```
### Meta Extras
```bash
pip install roomkit[providers] # All AI + transport providers
pip install roomkit[sources] # All event-driven sources
pip install roomkit[all] # Everything installable together
pip install roomkit[dev] # Development (test + lint + type check)
```
### Incompatible Extras
`qwen-tts`, `qwen-asr` and `neutts` each pin an exact `transformers` version
(4.57.3, 4.57.6 and 5.1.0), so they cannot be installed together. They are also
incompatible with `smart-turn`, whose floor is `transformers>=5.5` — the first
version without GHSA-fgcw-684q-jj6r — and therefore with `all`, which includes
`smart-turn`.
```bash
pip install roomkit[all] # includes smart-turn
pip install roomkit[qwen-tts] # in its own environment
```
`pyproject.toml` declares these conflicts, so an unsatisfiable combination fails
at resolution naming the two extras rather than backtracking onto an older
`transformers`.
## Environment Variables
Provider-specific API keys are passed via configuration objects, not environment variables. Example:
```python
from roomkit.providers.anthropic.config import AnthropicConfig
config = AnthropicConfig(api_key="sk-ant-...", model="claude-opus-5")
```
For voice providers, use lazy loaders to avoid import-time dependency checks:
```python
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
```bash
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.
```python
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
1. **RoomKit()** created the framework with in-memory defaults (store, locks, realtime).
2. **Channels** were registered globally, then **attached** to a room.
3. AI channel was attached with `category=ChannelCategory.INTELLIGENCE` — it receives messages and generates responses.
4. A **BEFORE_BROADCAST** hook runs synchronously before every broadcast. It can `block()`, `allow()`, or `modify()` events.
5. **process_inbound()** ran the full pipeline: route -> parse -> identity -> hooks -> store -> broadcast.
6. 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:
```python
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:
```
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.
### Minimal Channel→AI Example
```python
from roomkit import RoomKit, AIChannel, WebSocketChannel, ChannelCategory, InboundMessage, TextContent
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-opus-5",
)))
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!"))
)
```
### Pipeline vs Pipeline
RoomKit uses "Pipeline" in two contexts:
| 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:
```python
from roomkit import RoomKit, SQLiteStore
from roomkit.store.postgres import PostgresStore
embedded = RoomKit(store=SQLiteStore("roomkit.db"))
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:
1. **VoiceBackend** — Pure audio transport (mic, SIP, RTP, WebRTC). No speech detection.
2. **AudioPipeline** — Processes audio frames: Resampler -> Recorder -> AEC -> AGC -> Denoiser -> VAD -> Diarization.
3. **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
```python
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
```python
from roomkit import RoomKit
kit = RoomKit()
# Create room
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
The `ChannelType` enum defines 23 values: `SMS`, `MMS`, `RCS`, `EMAIL`, `WHATSAPP`, `WHATSAPP_PERSONAL`, `WEBSOCKET`, `AI`, `VOICE`, `REALTIME_VOICE`, `REALTIME_AUDIO_VIDEO`, `PUSH`, `MESSENGER`, `TELEGRAM`, `TEAMS`, `DISCORD`, `BUZZ`, `WEBHOOK`, `VIDEO`, `AUDIO_VIDEO`, `CONFERENCE`, `CLI`, `SYSTEM`. RoomKit ships with these factories and channel classes:
| Channel | Factory / Class | Category | Use Case |
|---------|----------------|----------|----------|
| SMS | `SMSChannel(id, provider=...)` | TRANSPORT | Text messages via Twilio, Telnyx, Sinch, VoiceMeUp |
| RCS | `RCSChannel(id, provider=..., fallback=True)` | 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 |
| Discord | `DiscordChannel(id, provider=...)` | TRANSPORT | Discord bot (recipient key `discord_channel_id`) |
| Buzz | `BuzzChannel(id, provider=...)` | TRANSPORT | Nostr relay channels (recipient key `buzz_channel_id`) |
| Webhook | `HTTPChannel(id, provider=...)` | TRANSPORT | Generic HTTP webhook (`ChannelType.WEBHOOK`) |
| WebSocket | `WebSocketChannel(id)` | TRANSPORT | Real-time bidirectional |
| CLI | `CLIChannel(id)` | TRANSPORT | Interactive terminal (stdin/stdout REPL) |
| AI | `AIChannel(id, provider=...)` | INTELLIGENCE | LLM responses |
| Agent | `Agent(id, provider=...)` | INTELLIGENCE | Agent with tools + greeting |
| ACP | `ACPChannel(id, command, cwd=...)` | INTELLIGENCE | External coding agent via Agent Client Protocol (`channel_type` is `AI`; one session per room) |
| Voice | `VoiceChannel(id, stt=..., tts=..., backend=...)` | TRANSPORT | Real-time audio |
| Realtime Voice | `RealtimeVoiceChannel(id, provider=...)` | TRANSPORT | Speech-to-speech AI |
| Video | `VideoChannel(id, backend=..., vision=...)` | TRANSPORT | Video track with vision analysis |
| Audio+Video | `AudioVideoChannel(id, backend=..., stt=..., tts=...)` | TRANSPORT | Combined voice + video |
| Realtime Audio+Video | `RealtimeAudioVideoChannel(id, provider=..., transport=...)` | TRANSPORT | Speech-to-speech AI with video |
| Conference | `ConferenceChannel(id, backend=...)` | TRANSPORT | Multi-party SFU conference with AI bot (LiveKit or mock backend) |
`MMS`, `PUSH`, and `SYSTEM` have no factory: `MMS` is set automatically when an SMS channel carries media, `SYSTEM` marks framework-generated events, and `PUSH` is reserved.
## Registering and Attaching Channels
Channels are registered globally, then attached to specific rooms:
```python
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-opus-5",
)))
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:
```python
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.
```python
# 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:
```python
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 |
```python
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:", format="markdown", plain_text="Choose:", buttons=[{"text": "A", "payload": "a"}])
# Media, Audio, Location, Template
MediaContent(url="https://example.com/image.png", mime_type="image/png", caption="Photo")
AudioContent(url="https://example.com/voice.ogg", mime_type="audio/ogg", transcript="Hello there")
LocationContent(latitude=40.7128, longitude=-74.0060, label="NYC Office")
TemplateContent(template_id="order_confirm", language="en", parameters={"1": "ORD-123"})
# Multi-part (max depth 5)
CompositeContent(parts=[
TextContent(body="Here's the photo:"),
MediaContent(url="https://example.com/photo.jpg", mime_type="image/jpeg"),
])
# Edit and Delete
EditContent(target_event_id="evt-abc", new_content=TextContent(body="Corrected text"))
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
1. Event broadcast starts → EventRouter iterates over all target channel bindings
2. For each target, call `ContentTranscoder.transcode(content, source_binding, target_binding)`
3. Transcoder checks `target_binding.capabilities.media_types` to decide if content passes through or needs conversion
4. If transcode returns `None` → delivery is skipped for that channel
### Channel Capabilities
Each channel binding declares what content types it supports via `ChannelCapabilities`:
```python
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,
)
```
`ChannelMediaType` enum: `TEXT`, `RICH`, `MEDIA`, `AUDIO`, `VIDEO`, `LOCATION`, `TEMPLATE`.
### Default Fallback Chain
| 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]")` |
### Multichannel Transcoding Example
A single event gets adapted differently for each channel:
```python
# 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]")
```
### Custom Content Transcoder
Implement the `ContentTranscoder` ABC to customize how content is adapted:
```python
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] + "...")
return content
# Override the default transcoder on the RoomKit instance
kit = RoomKit()
kit._transcoder = MyTranscoder()
```
## Detaching Channels
```python
await kit.detach_channel("room", "channel-id")
```
## Binding Metadata Updates
```python
await kit.update_binding_metadata("room", "ai-agent", {"temperature": 0.9})
```
## Querying Rooms
```python
# 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)
```
A participant is one shared record per `(room_id, participant_id)`, even when
`connected_via` names several channels. Its `status` is therefore global to the
record, not per-channel presence. When two channel memberships have independent
connect/disconnect lifecycles, give them distinct participant ids and correlate
them through `identity_id`.
---
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
```python
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`:
```python
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)
```
### What happens when a hook fails
A hook that does not produce a usable result — it raises, it exceeds its
timeout, it returns something that is not a `HookResult`, or it returns a
`modify` whose payload is the wrong type for the trigger — is treated as
**allow**, and the error is logged. A broken hook must not be able to take a
room down, so the event goes through.
**Two triggers invert that**, because their payload is content a hook may
exist to withhold:
| Trigger | On hook failure |
|---|---|
| `BEFORE_TTS` | **Blocked** — the text is not synthesised |
| `ON_TRANSCRIPTION` | **Blocked** — the transcript is not published |
| every other sync trigger | Allowed, error logged |
All four failure modes block on those two, not just exceptions: a rule that
covered exceptions but allowed timeouts would leak through the timeout.
The consequence for `BEFORE_BROADCAST` is worth being explicit about, since
it is the trigger most often used for moderation: **a moderation hook that
crashes lets the content through.** That is deliberate (RFC §9.3), not an
oversight. Two levers exist if you need more, and their exact scope matters:
- The `hook_error` framework event is emitted by the **inbound pipeline's**
`BEFORE_BROADCAST` pass only. Other sync-hook passes (`ON_TRANSCRIPTION`,
`BEFORE_TTS`, the re-entry path) and every async trigger report failures
in the log, not through `hook_error` — do not build monitoring for those
on this event.
- To make additional triggers fail closed, extend the set the engine reads
through its instance — RoomKit builds its own `HookEngine`, so there is
no constructor to subclass into:
```python
kit.hook_engine.FAIL_CLOSED_TRIGGERS = kit.hook_engine.FAIL_CLOSED_TRIGGERS | {
HookTrigger.BEFORE_BROADCAST,
}
```
Weigh it first: fail-closed moderation means an outage in your hook is an
outage of the room.
## Hook Priority
Lower priority numbers run first. Default is 0.
```python
# 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:
```python
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:
```python
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
All 76 `HookTrigger` enum values (`src/roomkit/models/enums.py`). **Execution** is how
the engine invokes the trigger: SYNC triggers run through the sequential,
priority-ordered sync pipeline (can block/modify); ASYNC triggers run
concurrently, fire-and-forget, errors logged. Registration mode is forgiving in
both directions: ASYNC-registered hooks on a SYNC trigger fire as observers
after the sync pass; SYNC-registered hooks (the default) on an ASYNC trigger
fire like any other observer. Triggers marked *Reserved* exist in the enum but
are not fired by any built-in code.
### 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. |
### Event Mutation (edit/delete)
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_EVENT_UPDATED` | ASYNC | `(event, ctx) -> None` | A persisted event's stored state changed — inbound `EditContent` or `kit.update_event()`. Payload is the updated event (`metadata.edited=True` on the edit path). |
| `ON_EVENT_DELETED` | ASYNC | `(event, ctx) -> None` | A persisted event was deleted — inbound `DeleteContent` (soft, `metadata.deleted=True`) or `kit.delete_event()` (hard; payload is the pre-delete snapshot). |
### 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 |
### Membership
Synthetic system events (`SystemContent`, `visibility=INTERNAL`), fired by the
member management API.
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PARTICIPANT_JOINED` | ASYNC | `(event, ctx) -> None` | Member added via `kit.add_member()`. `content.data`: `participant_id`, `identity_id`. |
| `ON_PARTICIPANT_LEFT` | ASYNC | `(event, ctx) -> None` | Member removed via `kit.remove_member()` — soft status flip to LEFT/BANNED. `content.data`: `participant_id`, `status`. |
| `ON_PARTICIPANT_UPDATED` | ASYNC | `(event, ctx) -> None` | Member renamed via `kit.rename_member()` (display name only; identity never changes). |
### Delivery
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_DELIVERY_STATUS` | ASYNC | `(DeliveryStatus, ctx) -> None` | Provider delivery receipt dispatched by `kit.process_delivery_status()` / `kit.process_webhook()`. The `@kit.on_delivery_status` decorator registers a `(status)`-only callback on this trigger. |
| `BEFORE_DELIVER` | ASYNC | `(event, ctx) -> None` | Before a proactive `kit.deliver()` strategy executes (in-process and worker paths). Payload is a synthetic INTERNAL system-source event describing the delivery. Observational — invoked through the async pipeline, cannot block (RFC §9 lists it SYNC; code invokes it ASYNC). |
| `AFTER_DELIVER` | ASYNC | `(event, ctx) -> None` | After the delivery strategy completes. Same payload shape with status DELIVERED/FAILED and `metadata.error` set on failure. |
### Side Effects
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_TASK_CREATED` | ASYNC | `(event, ctx) -> None` | A `Task` returned in `HookResult.tasks` was persisted. Payload event has type `TASK_CREATED` and `metadata.task_id`/`task_title`. Fires once per persisted task. |
| `ON_ERROR` | ASYNC | `(event, ctx) -> None` | Error during processing (every provider/inference failure path funnels here). Payload event carries `metadata.error`, `error_type`, `error_category`. |
### Session Lifecycle
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_SESSION_STARTED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | A session began: voice session bound, realtime session opened, conference bot connected (`event.session` = the session/bot), or first inbound on an auto-created text room. On the inbound path, internal (`_`-prefixed) hooks are awaited (greeting gate ordering) and user hooks fire in the background. Auto-greeting is an internal hook on this trigger. |
### Voice
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_SPEECH_START` | ASYNC | `(VoiceSession, ctx) -> None` | VAD detected speech start (VoiceChannel, RealtimeVoiceChannel). Conference lanes fire it per track with a synthetic system event. |
| `ON_SPEECH_END` | ASYNC | `(VoiceSession, ctx) -> None` | VAD detected speech end. Same per-lane conference behavior. |
| `ON_TRANSCRIPTION` | SYNC | `(event, ctx) -> HookResult` | STT produced a final transcript. Can block/modify; **fails closed** — a hook that raises/times out blocks publication. Payload: `TranscriptionEvent` (VoiceChannel), `RealtimeTranscriptionEvent` (RealtimeVoiceChannel), `ConferenceTranscription` (ConferenceChannel). Modify by returning the same type (voice/realtime also accept a plain `str`) with the new text. |
| `ON_PARTIAL_TRANSCRIPTION` | ASYNC | `(PartialTranscriptionEvent, ctx) -> None` | Streaming interim STT result. Hot path — skipped entirely when no hooks are registered. |
| `BEFORE_TTS` | SYNC | `(text: str, ctx) -> HookResult` | Before text is synthesized. Payload is the text `str`; modify with a `str`. Can block; **fails closed**. |
| `AFTER_TTS` | ASYNC | `(text: str, ctx) -> None` | After TTS audio was sent. Payload is the final synthesized text. |
### Voice Pipeline
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_VAD_SILENCE` | ASYNC | `(VADSilenceEvent, ctx) -> None` | VAD detected silence |
| `ON_VAD_AUDIO_LEVEL` | ASYNC | `(VADAudioLevelEvent, ctx) -> None` | Audio level update from VAD (high-frequency; telemetry-suppressed) |
| `ON_SPEAKER_CHANGE` | ASYNC | `(SpeakerChangeEvent, ctx) -> None` | Diarization detected speaker change |
| `ON_BARGE_IN` | ASYNC | `(BargeInEvent, ctx) -> None` | User interrupted TTS playback; carries `interrupted_text`, `audio_position_ms` |
| `ON_TTS_CANCELLED` | ASYNC | `(TTSCancelledEvent, ctx) -> None` | TTS playback was cancelled (barge-in or explicit interrupt) |
| `ON_DTMF` | ASYNC | `(DTMFDetectedEvent, ctx) -> None` | DTMF tone detected |
| `ON_TURN_COMPLETE` | ASYNC | `(TurnCompleteEvent, ctx) -> None` | Turn detector says turn is complete; carries combined text + confidence |
| `ON_TURN_INCOMPLETE` | ASYNC | `(TurnIncompleteEvent, ctx) -> None` | Turn detector says turn is incomplete |
| `ON_BACKCHANNEL` | ASYNC | `(BackchannelEvent, ctx) -> None` | Backchannel detected (uh-huh, yeah) |
| `ON_RECORDING_STARTED` | ASYNC | `(RecordingStartedEvent, ctx) -> None` | Audio recording started (voice session or conference track) |
| `ON_RECORDING_STOPPED` | ASYNC | `(RecordingStoppedEvent, ctx) -> None` | Audio recording stopped, result available |
| `ON_INPUT_AUDIO_LEVEL` | ASYNC | `(AudioLevelEvent, ctx) -> None` | Inbound audio level (throttled; telemetry-suppressed) |
| `ON_OUTPUT_AUDIO_LEVEL` | ASYNC | `(AudioLevelEvent, ctx) -> None` | Outbound audio level (throttled; telemetry-suppressed) |
| `BEFORE_BRIDGE_AUDIO` | SYNC | `(BridgeAudioEvent, ctx) -> HookResult` | Before an audio frame is forwarded across an audio bridge. Can block/modify the frame. Only invoked when hooks are registered — otherwise frames bypass the event loop for latency. |
### Tool Execution
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_TOOL_USE` | SYNC | `(ToolCallEvent, ctx) -> HookResult` | Before a tool executes (AIChannel, realtime channels, external/ACP tools). Block denies the call. Fails closed: if room context cannot be built, the call is denied. |
| `ON_TOOL_CALL` | SYNC | `(ToolCallEvent, ctx) -> HookResult` | A tool call executed (AIChannel, RealtimeVoiceChannel, skills, external/ACP tools). Block returns an error result to the model; `HookResult(metadata={"result": ...})` supplies or overrides the tool result (`event.result is None` means the hook must provide it). RFC §9 lists this ASYNC; code invokes it through the sync pipeline. |
| `ON_USER_INPUT_REQUIRED` | SYNC | `(PendingInputEvent, ctx) -> HookResult` | A `HumanInputHandler`-backed tool paused awaiting human input. Sync so the notification (e.g. WebSocket push) lands before `wait()` starts blocking; block auto-rejects the pending request. |
### Realtime Voice
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_REALTIME_TEXT_INJECTED` | ASYNC | `(event, ctx) -> None` | Text was injected into realtime session |
### AI Generation
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_AI_GENERATION` | SYNC | `(AIGenerationEvent, ctx) -> HookResult` | Before the AI provider is invoked. `event.ai_context` (messages, system prompt, tools) may be mutated in place; block skips generation. |
| `ON_AI_THINKING` | ASYNC | `(ThinkingEvent, ctx) -> None` | A reasoning window opened (`THINKING_START`) on an AIChannel; fires whether or not a realtime backend is configured. The reasoning text itself streams on the ephemeral `THINKING_DELTA` / `THINKING_END` events, so `event.thinking` is empty here. |
| `ON_AI_RESPONSE` | ASYNC | `(AIResponseEvent, ctx) -> None` | AI generation completed. Carries the turn's text — `response_content`, the segments a tool call cut joined with a blank line, and `segments` one by one — plus usage, latency, tool call count, and `loop_end_reason` — which of the tool loop's rules ended the turn (`completed`, `max_rounds`, `timeout`, `cancelled`, `force_stopped`, `truncated`, `empty_response`). Read that rather than inferring a cut-off from `tool_calls_count`, which counts the calls the turn ran. |
### Protocol Observability
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PROTOCOL_TRACE` | ASYNC | `(ProtocolTrace, ctx) -> None` | Transport-level protocol trace (SIP, RTP, …) forwarded from a channel. Traces arriving before the room exists are buffered and replayed on attach. |
### Orchestration (multi-agent)
Handoff triggers fire from the `HandoffCoordinator` with a synthetic INTERNAL
system event (AI-channel source; `metadata`: `from_agent`, `to_agent`,
`accepted`, `new_phase`).
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_HANDOFF` | ASYNC | `(event, ctx) -> None` | Agent handoff accepted and executed. |
| `ON_HANDOFF_REJECTED` | ASYNC | `(event, ctx) -> None` | Agent handoff was rejected (`metadata.accepted=False`). |
| `ON_PHASE_TRANSITION` | ASYNC | `(event, ctx) -> None` | Fired alongside `ON_HANDOFF` after an accepted handoff, carrying the new phase. |
| `ON_STATUS_POSTED` | ASYNC | `(StatusEntry, ctx) -> None` | A `StatusBus` post whose `metadata` carries a `room_id` ran that room's hooks; a post with no room reaches no hook (the bus is global). Every post also emits the `status_posted` framework event. |
### Delegation (background tasks)
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_TASK_DELEGATED` | ASYNC | `(event, ctx) -> None` | `kit.delegate()` dispatched a task to a child room. INTERNAL event, type `TASK_DELEGATED`; `metadata`: `task_id`, `child_room_id`, `agent_id`. |
| `ON_TASK_COMPLETED` | ASYNC | `(event, ctx) -> None` | A delegated task finished. Fires in the parent room; type `TASK_COMPLETED`, body = output or error. |
### Video
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `BEFORE_BRIDGE_VIDEO` | SYNC | `(BridgeVideoEvent, ctx) -> HookResult` | Before a video frame is forwarded across a bridge. Can block/modify the frame. Only invoked when hooks are registered (fast path bypasses the event loop). |
| `ON_VIDEO_SESSION_STARTED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | Video path live (VideoChannel, AVChannel, RealtimeAVChannel). `event.session` is the Video/VoiceSession. |
| `ON_VIDEO_SESSION_ENDED` | ASYNC | `(SessionStartedEvent, ctx) -> None` | Video session ended (same payload shape). |
| `ON_VIDEO_TRACK_ADDED` | — | — | Reserved — defined in the enum, not fired by built-in channels. |
| `ON_VIDEO_TRACK_REMOVED` | — | — | Reserved — defined in the enum, not fired by built-in channels. |
| `ON_VISION_RESULT` | SYNC | `(VisionEvent, ctx) -> HookResult` | A VisionProvider analyzed a frame. Block discards the result; modify rewrites the description before event injection and AI context update. RFC §9 lists this ASYNC; code invokes it through the sync pipeline. |
| `ON_SCREEN_SHARE_STARTED` | ASYNC | `(event, ctx) -> None` | ConferenceChannel: a SCREEN_SHARE track was published. `content.data`: `track_id`, `participant_id`. |
| `ON_SCREEN_SHARE_STOPPED` | ASYNC | `(event, ctx) -> None` | ConferenceChannel: a SCREEN_SHARE track was unpublished. |
| `ON_VIDEO_DETECTION` | ASYNC | `(VideoDetectionEvent, ctx) -> None` | Video pipeline filter emitted a detection event (object, face, …). |
### Conference (SFU)
All fired by `ConferenceChannel` as synthetic system events
(`SystemContent` with a `data` dict including `channel_id`); the channel's own
bot never triggers them.
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_CONFERENCE_PARTICIPANT_JOINED` | ASYNC | `(event, ctx) -> None` | Participant joined the conference media session. `data`: `participant_id`. |
| `ON_CONFERENCE_PARTICIPANT_LEFT` | ASYNC | `(event, ctx) -> None` | Participant left the conference media session. |
| `ON_CONFERENCE_TRACK_PUBLISHED` | ASYNC | `(event, ctx) -> None` | Participant published a track. `data`: `track_id`, `participant_id`, `kind` (audio/video/screen_share). |
| `ON_CONFERENCE_TRACK_UNPUBLISHED` | ASYNC | `(event, ctx) -> None` | A track was unpublished (same `data` shape). |
| `ON_CONFERENCE_TRACK_MUTED` | ASYNC | `(event, ctx) -> None` | Publisher muted a track — "camera off" is usually a muted VIDEO track, not an unpublish. |
| `ON_CONFERENCE_TRACK_UNMUTED` | ASYNC | `(event, ctx) -> None` | Publisher unmuted a track. |
| `ON_ACTIVE_SPEAKER_CHANGED` | ASYNC | `(event, ctx) -> None` | SFU reported a dominant-speaker change. `data`: `participant_id`. |
| `ON_CONNECTION_QUALITY_CHANGED` | ASYNC | `(event, ctx) -> None` | SFU reported a participant's connection quality. `data`: `participant_id`, `quality`. |
### Planning
| Trigger | Execution | Signature | Description |
|---------|-----------|-----------|-------------|
| `ON_PLAN_UPDATED` | ASYNC | `(PlanUpdatedEvent, ctx) -> None` | The `plan_tasks` tool (`enable_planning=True`) updated the room's task plan; `event.tasks` carries the new list. The same update also goes out as an ephemeral realtime event `{"type": "plan_updated"}` for live UIs. |
### Feedback
| 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):
```python
@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
```python
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-opus-5",
)),
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]` |
| Gemini on Vertex AI | `GeminiVertexProvider` | `GeminiVertexConfig` | `roomkit[gemini]` |
| Mistral | `MistralAIProvider` | `MistralConfig` | `roomkit[mistral]` |
| Azure OpenAI | `AzureAIProvider` | `AzureAIConfig` | `roomkit[azure]` |
| OpenRouter (300+ models) | `OpenRouterAIProvider` | `OpenRouterConfig` | `roomkit[openrouter]` |
| LiteLLM proxy (self-hosted gateway) | `LiteLLMAIProvider` | `LiteLLMConfig` | `roomkit[litellm]` |
| xAI (Grok) | `XAIAIProvider` | `XAIConfig` | `roomkit[xai]` |
| PolarGrid (Canadian-hosted) | `PolarGridAIProvider` | `PolarGridConfig` | `roomkit[polargrid]` |
| vLLM (local) | `create_vllm_provider()` | `VLLMConfig` | `roomkit[vllm]` |
| Ollama (local) | `OllamaAIProvider` | `OllamaConfig` | `roomkit[ollama]` |
| Mock (testing) | `MockAIProvider` | — | built-in |
Provider notes:
- **Explicit model selection** — `model=` is required by `OpenAIConfig` and
`AnthropicConfig`; upgrading RoomKit therefore cannot silently change cost,
latency, or model behavior. For the selected model, `OpenAIConfig`
automatically uses `max_completion_tokens` and omits custom temperature for
current GPT-5 and o-series ids, while `AnthropicConfig` uses adaptive thinking
and omits temperature for current Claude reasoning ids. Explicit flags take
precedence, and a custom `base_url` keeps conservative legacy behavior.
- **Gemini on Vertex AI** (`roomkit.providers.gemini.vertex`) — subclass of `GeminiAIProvider` serving the same Gemini models through a Google Cloud project with a pinned region (`GeminiVertexConfig` requires `project` and `location`, e.g. `"northamerica-northeast1"`; no API key — the identity is `impersonate_service_account`, else `service_account_json`, else ADC). Use it when data residency matters (Québec Law 25 / PIPEDA). Generation, streaming, thinking, and the model catalog are inherited unchanged.
- **OpenRouter** (`roomkit.providers.openrouter`) — subclass of `OpenAIAIProvider` pointed at `https://openrouter.ai/api/v1`; `OpenRouterConfig` subclasses `OpenAIConfig` (adds `site_url`/`app_name` attribution headers) and `model` is a required slug like `"anthropic/claude-sonnet-4.5"`. Reasoning is forwarded to any upstream model via OpenRouter's unified `reasoning` object. Model-listing nuance: OpenRouter's `/models` items omit the `object`/`owned_by` fields the OpenAI SDK expects, so `list_models()` reads the raw JSON instead.
- **LiteLLM proxy** (`roomkit.providers.litellm`) — subclass of `OpenAIAIProvider` pointed at a self-hosted LiteLLM gateway (default `http://localhost:4000`); the extra installs the `openai` SDK, deliberately **not** the `litellm` package (the gateway keeps keys/budgets/routing server-side). `model` is the deployment's public alias and `api_key` a virtual or master key, both required. Reasoning rides LiteLLM's cross-provider normalisation: `reasoning_effort` passes through, `thinking_budget>0` maps to a `thinking` token budget, and the trace comes back in `reasoning_content`; `thinking_budget=0` sends no reasoning params at all (LiteLLM has no disable token every upstream translator accepts — force-off belongs in the proxy's per-model config or `extra_body`). `available_models()` is empty (the model list is the operator's config); `list_models()` reads the proxy's `/model/info` — context window, vision flag, and per-token costs per alias, with a load-balanced group merged into the one entry every deployment can honour (smallest window, vision and price only when unanimous) and LiteLLM's `0`-for-unknown costs mapped to "unpriced" rather than free.
- **xAI (Grok)** (`roomkit.providers.xai`) — subclass of `OpenAIAIProvider` pointed at `https://api.x.ai/v1`; defaults to `max_completion_tokens` and stream usage. `XAIRealtimeProvider` (Grok speech-to-speech) is a separate import from `roomkit.providers.xai.realtime`.
- **PolarGrid** (`roomkit.providers.polargrid`) — Canadian-hosted inference network (edges in Toronto, Vancouver, Montréal) via the official `polargrid-sdk` async client; OpenAI-shaped chat-completions surface. Supports tool calling, thinking (`PolarGridConfig(thinking=True)` sets the `enable_thinking` request flag; reasoning is surfaced as `AIResponse.thinking` / `StreamThinkingDelta`), and vision (`image_url` content parts). Pin `region` in production when residency matters.
Pick **Ollama** over the OpenAI-compat shim (`OpenAIAIProvider` pointed at `http://host:11434/v1` or `create_vllm_provider()` with an Ollama URL) whenever the model is a reasoning model (DeepSeek-R1, Qwen 3 thinking variants, etc.) — only the native API exposes the `think` parameter and streams the `thinking` field separately from `content`. See `docs/c7/ollama-provider.md` for the full rundown.
```python
# 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?"])
```
## Model Catalog and Pricing
`provider.catalog_entry()` returns the configured model's offline `ModelInfo`.
When it carries `pricing`, `pricing.cost_for(response.usage)` prices fresh
input, output, cache reads and represented cache writes. A `None` cache rate
means no separate per-token charge is represented; it contributes zero rather
than falling back implicitly. Catalogs repeat the input rate explicitly when a
vendor bills a cache counter as ordinary input.
Tiered entries also carry a long-context threshold plus input/output
multipliers. `cost_for()` applies them automatically to GPT-5.6, Gemini Pro and
current Grok usage after total input crosses the vendor's threshold.
## Agent Class
`Agent` extends `AIChannel` with role, description, greeting, and memory support — designed for multi-agent orchestration:
```python
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:
```python
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:
```python
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:
```python
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:
```python
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())
```
## Tool Search (Progressive Tool Disclosure)
When an agent has dozens of tools, sending every schema to the model on
every turn burns context and makes smaller models hallucinate tool names.
**Tool Search** hides the catalogue behind two discovery tools and lets the
model reveal only what it needs:
- `find_tools(query)` — search the catalogue by natural language; the
matches become directly invocable for the rest of the turn.
- `list_tools(category=None)` — list the catalogue (name + short description).
```python
ai = AIChannel(
"ai",
provider=provider,
tool_handler=handle_tools,
tools=big_catalogue, # e.g. 60+ MCP tools
tool_search=None, # None = auto, True/False = force
tool_search_threshold_pct=10.0, # auto-enable above this % of the window
tool_search_threshold=20, # fallback tool count when window unknown
tool_search_pinned=["get_help"], # always visible, never searched for
)
```
How it works:
1. The model first sees only `find_tools`/`list_tools` plus the pinned set —
the discretionary catalogue is hidden.
2. It calls `find_tools("send a text message")`; the matches are scored and
returned, and their names are recorded for the turn.
3. On the **next** tool-loop round the matched tools are visible and directly
callable. The text loop re-sends its (re-filtered) tool list every round,
so no provider reconfigure is needed — this works on **any** text/HTTP
provider. (The realtime voice channel offers the same feature via
`provider.reconfigure`.)
Notes:
- **Activation auto-tunes to the model.** In `auto` mode it defers when the
*deferrable* (non-pinned) tools would cost more than `tool_search_threshold_pct`
% of the model's context window (default 10%) — a large model is a no-op, a
small one defers early. When the window is unknown (custom / local model ids
absent from the provider catalog) it falls back to the `tool_search_threshold`
tool count. Below the threshold Tool Search is a no-op and every tool is sent.
- **Pinned tools** stay visible without a search. The discovery tools always
pass tool-policy and skill gating, so they work even under a restrictive
`tool_policy`.
- A second `find_tools` call **swaps** the revealed window (keeping the visible
surface small); `list_tools` reveals nothing — it is purely informational.
See `examples/ai_tool_search.py` for a runnable, no-API-key walkthrough.
## Agent Skills
A skill is a directory holding a `SKILL.md` (YAML frontmatter + an instructions
body) and optional `scripts/` and `references/` — the
[Agent Skills](https://agentskills.io) standard. It packages knowledge the model
loads **on demand** instead of knowledge every system prompt has to carry.
```python
from roomkit.skills import SkillRegistry
registry = SkillRegistry()
count = registry.discover("./skills") # returns how many were found
ai = AIChannel(
"ai",
provider=provider,
skills=registry,
skills_in_prompt=True, # False = the host renders its own manifest
script_executor=my_executor, # omit and run_skill_script is not offered
)
```
`discover()` takes one or more directories and commits only once every
candidate has parsed, so a failure leaves the registry as it was rather than
half filled. It raises on a malformed skill by default (`strict=False` skips it):
a bad skill is a deployment error, and skipping it silently leaves an agent that
quietly cannot do something it was configured to do.
Three tools are registered automatically alongside the host's own:
| Tool | Offered | Effect |
|------|---------|--------|
| `activate_skill(name)` | always | Loads the skill for the conversation |
| `read_skill_reference(skill_name, filename)` | always | Reads one file from `references/` |
| `run_skill_script(skill_name, script_name, arguments)` | only with `script_executor` | Runs a script through the integrator's executor |
### Activation lifecycle
**An activation lasts the conversation, not the turn that made it.** Re-sending
a 9 KB body on every turn costs more than the skill is worth, so the body moves
to where the per-turn rebuild carries it for free:
| | First `activate_skill` in a room | Later calls |
|---|---|---|
| Tool result | Full `instructions` + the `scripts`/`references` listing | `{"ok": true, "already_active": true, ...}` — no body |
| System prompt | *(nothing yet — composed before the call)* | `# Active skill instructions (binding rules)` carrying the body |
| Gated tools | Revealed for the rest of the turn | Still revealed, no re-activation |
- The ack is safe because the prompt carries the rules. Lose the record (process
restart, channel object replaced) and the prompt block goes with it, so the
next activation returns the body again — it degrades to reloading, never to an
ack with no rules.
- The record is **hydrated** from the room's persisted tool-call history, so a
channel swapped mid-conversation does not restart amnesic.
- **Bodies are never evicted.** Large tool results are normally replaced by a
`read_stored_result` pointer; an `activate_skill` result is exempt, because
binding rules cut to a head/tail preview are not rules. References still
evict — those are data, and paginating data is what eviction is for.
- **Four skills stay active per room**, by recency; a fifth retires the least
recently used one, whose next `activate_skill` returns the body again.
### Reading the active set
`skills_in_prompt=False` hands the manifest to the host, and a catalogue is only
half of what a manifest needs. `active_skill_names(room_id)` supplies the other
half — the skills that room is already carrying:
```python
active = ai.active_skill_names("support-room") # {"code-review"}
rows = [
f"- {m.name} ({'loaded' if m.name in active else 'available'}): {m.description}"
for m in registry.all_metadata()
]
```
Without it every row reads *available*, including the skill whose instructions
the prompt already carries, so the manifest asks the model to load rules that
are in front of it — one wasted round, answered by an ack. Keyed on the room,
empty for a room that activated nothing and for `None`. Active bodies are
injected whether or not `skills_in_prompt` is set: that is runtime state a host
cannot know.
### Tool gating
A skill's `allowed_tools` frontmatter (a YAML list or a comma-separated scalar)
names the tools it unlocks. Entries are `ToolPolicy` **globs** (RFC §24.2), so
`search_*` covers every tool whose name starts with `search_` — match them, never
test membership. A gated tool is hidden from the catalogue *and* refused at
execution, because a model that saw the name before the skill was deactivated can
still call it. The skill and Tool Search infrastructure tools are exempt from
gating in both places: gating `find_tools` or `activate_skill` would tell the
model to activate a skill it has no way left to name.
### Visibility states
```python
registry.mark_unlisted("legacy-csv-import") # activatable, absent from the manifest
registry.mark_unavailable("deploy-helper", "needs a ScriptExecutor")
```
`registry.skill_names` is what can be activated, `registry.listed_names` what the
manifest advertises, and `registry.unavailable_skills` maps a name to the reason
the model can quote instead of guessing. `registry.to_prompt_xml()` renders the
manifest RoomKit would otherwise inject.
### Script execution
There is **no default executor** — sandboxing, timeouts and allowed interpreters
are the integrator's call. Implement `ScriptExecutor` and pass it as
`script_executor`. The script name comes from the model, so RoomKit resolves it
first: use `skill.resolve_script(name)` rather than joining
`skill.path / "scripts" / name` yourself. A name that escapes the skill —
including through a symlink planted in `scripts/` — raises `SkillPathError` and
never reaches your executor.
### Realtime voice
`RealtimeVoiceChannel` runs the same lifecycle per **session**, with
`skill_delivery_mode` deciding how a body reaches the model:
- `"on_demand"` — metadata only in the prompt; `activate_skill` loads the body
through `provider.reconfigure`.
- `"inline_full"` — every body baked into the initial `system_instruction`;
`activate_skill` becomes a declarative ack and no reconfigure is needed.
It defaults to `"inline_full"` when the provider reports
`supports_mid_session_reconfigure=False` (e.g. Gemini 3.x Flash Live), and to
`"on_demand"` otherwise.
Runnable, no API key needed for the last one: `examples/agent_skills.py`
(discovery and activation), `examples/skill_visibility.py` (the three states plus
a recommender hook), `examples/skill_active_manifest.py` (a host manifest reading
the active set).
## Per-Room Configuration
Override AI settings per room via binding metadata:
```python
await kit.attach_channel(
"billing-room",
"ai-agent",
category=ChannelCategory.INTELLIGENCE,
metadata={
"system_prompt": "You are a billing specialist.",
"temperature": 0.3,
"tools": [...],
},
)
```
## Per-Turn Configuration
Binding metadata is a snapshot taken at attach time. When the configuration
changes underneath you — admin edits, per-user gating, feature flags — that
snapshot becomes a second source of truth that goes stale. `config_provider`
resolves the config fresh at the start of every generation instead:
```python
from roomkit import AIChannel, AIChannelTurnConfig
async def per_turn(binding, context) -> AIChannelTurnConfig | None:
settings = await load_settings(context.room.id)
return AIChannelTurnConfig(
system_prompt=settings.prompt,
temperature=settings.temperature,
enable_thinking=settings.thinking,
reasoning_effort=settings.effort,
)
ai = AIChannel("ai-agent", provider=provider, config_provider=per_turn)
```
`AIChannelTurnConfig` (exported from `roomkit`) carries `system_prompt`,
`tools`, `temperature`, `max_tokens`, `thinking_budget`, `enable_thinking`
and `reasoning_effort`. Each setting resolves from the most specific source
that has an opinion: binding metadata (per-room operator intent, always
wins), then the `config_provider` result, then the `AIChannel` constructor
default, then the provider config. `None` at a tier means "not set here" and
defers outward, so an unset knob never overrides with a default.
## Streaming
AIChannel supports streaming responses to WebSocket clients:
```python
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}")
```
### Why a streaming tool loop stopped
The streaming tool loop ends on rules of its own — the round cap, the
wall-clock deadline, a round truncated at the output cap, a model that
answered nothing after its tools, the anti-loop ripcord, a cancellation. It
yields a final
`LoopEndMarker(reason, rounds)` on **every** exit, `completed` included, so
the end of the stream is never itself the signal and no consumer has to
re-derive the cause by counting tool calls and reading a clock.
Read it at the source, by subclassing `AIChannel` and wrapping
`ChannelOutput.response_stream`:
```python
from roomkit.channels.ai import AIChannel
from roomkit.models.streaming import LoopEndMarker
class ObservingAIChannel(AIChannel):
async def on_event(self, event, binding, context):
output = await super().on_event(event, binding, context)
if output.response_stream is None:
return output
return output.model_copy(
update={"response_stream": self._observe(output.response_stream)}
)
async def _observe(self, inner):
async for delta in inner:
if isinstance(delta, LoopEndMarker):
if delta.reason != "completed":
logger.warning(
"agent stopped: %s after %d rounds", delta.reason, delta.rounds
)
continue # keep the terminal marker out of the stream
yield delta
```
`reason` is one of `completed`, `max_rounds`, `timeout`, `truncated`,
`empty_response`, `force_stopped`, `cancelled`, `error`. The limits each
reason refers to are the caller's own configuration and are not repeated on
the marker.
`force_stopped` is the one worth special attention, because it is the only
non-`completed` exit that ends **with text**: the anti-loop guard pulled the
ripcord on a model re-issuing an already-blocked call, which strips the tools
and asks for a plain-text answer. That text summarises an interrupted turn, so
a consumer treating "the model produced prose" as "the model answered" will
deliver a cut run as a finished one — which is precisely why the reason is
named rather than folded into `completed`. `error` ends with text on the same
terms: a mid-loop provider failure salvaged as a partial answer (non-streaming
only — a streaming failure reaches the consumer as the exception itself).
### Why a non-streaming tool loop stopped
The non-streaming loop returns `RoomEvent`s rather than a stream, so its
reason travels on them: every response MESSAGE event's metadata carries
`loop_end_reason` with the same values, next to `ai_usage` (which sums every
generation round of the turn, not just the last).
```python
reply = output.response_events[-1]
if reply.metadata["loop_end_reason"] != "completed":
logger.warning("agent stopped: %s", reply.metadata["loop_end_reason"])
```
The framework's inbound streaming path forwards text deltas and the tool-call
and thinking markers to a channel's `deliver_stream`, but **not** the terminal
marker — it would reach a renderer as noise. Overriding `deliver_stream` on a
WebSocket or CLI channel will therefore not see it; wrap the AI channel's own
`response_stream` instead.
Additive by construction: the streaming protocol is a mixed
`str | StreamMarker` whose consumers already dispatch on the markers they
know, so a text-only consumer filtering on `isinstance(chunk, str)` is
unaffected. Streaming only — the non-streaming loop returns an `AIResponse`
the caller already holds.
Two bounds keep a degenerate model from running away with a turn:
`max_tool_rounds` caps how many rounds run, and a **32-call ceiling per
round** caps how wide one round may be. The per-round cap is applied before
the assistant message is assembled, so a dropped call is absent from the
transcript as well as from the results — no provider sees a tool call with no
matching result. Both loops enforce it from the shared loop rules.
## AI Thinking/Reasoning
Some providers support extended thinking:
```python
ai = AIChannel(
"ai",
provider=AnthropicAIProvider(AnthropicConfig(
api_key="...",
model="claude-opus-5",
)),
system_prompt="Think step by step.",
thinking_budget=4096, # Setting a budget enables thinking mode
)
```
Per-provider mechanisms: Anthropic uses `thinking_budget` as above; OpenAI-family providers (OpenAI, Azure, OpenRouter, LiteLLM, xAI) use the `reasoning_effort` config field — OpenRouter translates it into its unified `reasoning` object for any upstream model, and a LiteLLM gateway normalises it (plus an explicit `thinking` budget when `thinking_budget>0`) for every upstream it fronts; Gemini (and Vertex) use `GeminiConfig.thinking_level`; Ollama exposes the native `think` parameter (see `docs/c7/ollama-provider.md`); PolarGrid uses `PolarGridConfig(thinking=True)` (the `enable_thinking` flag, surfaced as `AIResponse.thinking` / `StreamThinkingDelta`).
`thinking_budget`, `enable_thinking` and `reasoning_effort` all ride the
per-turn chain described under *Per-Turn Configuration*, so reasoning is
steerable per room and per turn rather than only per provider instance:
```python
ai = AIChannel(
"ai",
provider=provider,
enable_thinking=True,
reasoning_effort="low",
)
```
A thinking model costs two to three times the tokens and the latency of a
direct answer, and that trade differs between an agent's tool loop — where the
model mostly shapes results it already has — and a chat turn where the
reasoning is the value.
Reasoning competes with the answer for the same `max_tokens`. A round that
spends its whole budget thinking returns empty `content` with a truncation
finish reason; RoomKit recognises that across every provider's spelling
(`length`, `max_tokens`, `MAX_TOKENS`, case-insensitively) and skips the
empty-response retry, which would only truncate again under the same cap.
`AIContext.max_tokens` defaults to `None`, not a number: every provider reads
`context.max_tokens or self._config.max_tokens`, so a non-`None` default here
would shadow the provider config and make a configured cap unreachable.
Providers whose config also defaults it to `None` (Ollama, PolarGrid) send no
cap at all, letting the server pick its own.
### vLLM reasoning and sampling
vLLM renders the model's chat template server-side, so reasoning is steered
through `chat_template_kwargs` rather than a top-level sampling parameter.
`VLLMConfig.enable_thinking` and `VLLMConfig.reasoning_effort` map onto them;
both default to `None`, leaving the model's own default (current Qwen builds
think at their most verbose effort unless told otherwise). An explicit
`extra_body["chat_template_kwargs"]` entry still wins.
`VLLMConfig` also types five sampling knobs that previously required
`extra_body`: `top_p`, `top_k`, `min_p`, `presence_penalty` and
`repetition_penalty`. Each defaults to `None` ("the server decides"); an
explicit `0` is sent, since `min_p=0.0` and `presence_penalty=0.0` are values
rather than absences.
```python
from roomkit.providers.vllm import VLLMConfig, create_vllm_provider
provider = create_vllm_provider(VLLMConfig(
model="Qwen/Qwen3-8B",
enable_thinking=False,
presence_penalty=1.5, # Qwen3's guidance for non-thinking mode
top_p=0.8,
top_k=20,
))
```
## Vision Support
AI providers that support vision can process images sent as `MediaContent`:
```python
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
```
OpenAI-family providers (OpenAI, Azure, OpenRouter, LiteLLM, xAI) and PolarGrid send images as OpenAI-shaped `image_url` content parts (remote URL or `data:` URI); PolarGrid and xAI gate this on the model's `supports_vision` from their curated catalogs — whether the model actually reads the image is the deployed model's capability.
## Multi-Speaker Rooms
AIChannel builds the model's history from the room's events, and every event that is not the AI's own becomes a `user` turn. In a room where several people speak — a Teams channel, a WhatsApp group, a shared inbox — that flattening erases who said what, and the model guesses the addressee wrong (a reply opening with the wrong colleague's name). So when the history window holds **two or more distinct speakers**, each attributable user turn reaches the model as `"Name: text"` and a one-line note is appended to the system prompt:
```text
Several people take part in this conversation. Their messages are prefixed with
the sender's name ("Name: message"). The prefix is transcript metadata, not text
they typed: rely on it to know who said what, and never prefix your own replies
with a name.
```
The speaker of an event is resolved in this order:
1. `event.metadata["sender_name"]` — stamped at ingress. The Teams webhook parser and the WhatsApp Personal (neonize) source write it; a host does the same by passing `InboundMessage(..., metadata={"sender_name": "Alice"})`.
2. The room's participant record: `Participant.display_name` for the participant whose `id` equals `event.source.participant_id` (the inbound `sender_id`).
3. Otherwise the turn has no speaker and is left bare, even in a multi-speaker room.
Rules that follow from the definition:
- Assistant turns are never prefixed — only `user` turns are.
- A multimodal user turn (`list[AITextPart | AIImagePart]`) gets a lead `AITextPart("Name:")` rather than a rewritten body.
- The trigger turn (the event the AI is replying to) is attributed like the history.
- The note is appended once per turn, after the channel's own `system_prompt`.
- A single-speaker room — a 1:1 DM — builds a byte-identical prompt: no prefixes, no note. Attribution is decided per turn from the window the memory provider returns, so it switches on the moment a second named speaker appears in it and off when the window no longer holds one.
There is nothing to configure. See `examples/ai_multi_speaker.py`.
## Agentic Features
### Dangling Tool Call Recovery
When a user sends a new message while the AI is mid-tool-execution (barge-in), tool calls can be left without matching results. AIChannel automatically detects these orphaned calls and injects synthetic cancellation results before the next AI turn, preventing provider API rejections.
This is fully automatic — no configuration needed.
### Large Output Eviction
When tool results are very large (database queries, file dumps, API responses), they consume significant context budget. AIChannel can evict large results to a side buffer and replace them with a preview:
```python
ai = AIChannel(
"ai-agent",
provider=provider,
system_prompt="You are a data analyst.",
evict_threshold_tokens=5000, # default: 5000 tokens
tools=[QueryDatabase()],
)
```
When a tool result exceeds the threshold:
1. The full result is stored in a FIFO-bounded buffer (max 50 entries)
2. A head/tail preview (first 5 + last 5 lines) replaces the result in context
3. A `_read_tool_result` tool is auto-injected so the agent can paginate through the full output
### Planning Tools
Enable structured task planning so agents can break down complex work and track progress:
```python
ai = AIChannel(
"ai-agent",
provider=provider,
system_prompt="You are a research assistant.",
enable_planning=True,
)
```
When enabled, the AI gets a `plan_tasks` tool that accepts up to 100 tasks with a title of at most 500 characters and a `status` (`pending`, `in_progress`, `completed`, `blocked`). Undeclared task fields are discarded. The current plan is:
- Injected into the system prompt on each turn (so the AI sees its progress)
- Published as an ephemeral `CUSTOM` event with `data.type = "plan_updated"` for real-time UI rendering
Subscribe to plan updates for UI:
```python
await kit.subscribe_room("room-1", my_callback)
# Callback receives ephemeral event with:
# type: "custom", data: {"type": "plan_updated", "tasks": [...]}
```
The ephemeral event is the only plan-update signal — the `HookTrigger.ON_PLAN_UPDATED` enum value is reserved and not currently fired.
### SummarizingMemory
For long conversations, use `SummarizingMemory` to proactively manage context budget with two tiers:
```python
from roomkit.memory import SummarizingMemory, SlidingWindowMemory
ai = AIChannel(
"ai-agent",
provider=main_provider,
memory=SummarizingMemory(
inner=SlidingWindowMemory(max_events=100),
provider=summary_provider, # lightweight model (e.g. Haiku)
max_context_tokens=128_000,
tier1_ratio=0.50, # truncate old events at 50%
tier2_ratio=0.85, # LLM summarization at 85%
),
)
```
- **Tier 1** (~50% capacity): Truncates large text bodies in older events to 2000 chars. No LLM call — cheap and fast.
- **Tier 2** (~85% capacity): Calls the summary provider to summarize older events into a concise paragraph. Keeps recent events at full fidelity. Supports chained summaries (prior summary is incorporated into the new one).
### Knowledge Retrieval (RAG)
Enrich AI context with external knowledge sources using `RetrievalMemory`:
```python
from roomkit.knowledge import KnowledgeSource, KnowledgeResult
from roomkit.memory import RetrievalMemory, SlidingWindowMemory
# Implement your own knowledge source (vector store, search engine, etc.)
class FAQSource(KnowledgeSource):
async def search(self, query, *, room_id=None, limit=5):
results = await my_vector_db.search(query, top_k=limit)
return [KnowledgeResult(content=r.text, score=r.score, source="faq") for r in results]
ai = AIChannel(
"ai-agent",
provider=provider,
memory=RetrievalMemory(
sources=[FAQSource()],
inner=SlidingWindowMemory(max_events=50),
max_results=5,
),
)
```
`RetrievalMemory` searches all sources concurrently, deduplicates results, and prepends relevant knowledge as a context message. When `ingest()` is called (automatic on every inbound event), it also indexes content in all sources.
#### Built-in: PostgreSQL Full-Text Search
For production use without a vector database, use `PostgresKnowledgeSource`:
```python
from roomkit.knowledge.postgres import PostgresKnowledgeSource
source = PostgresKnowledgeSource(dsn="postgresql://localhost/mydb")
await source.init()
# Or share the pool with PostgresStore:
source = PostgresKnowledgeSource(pool=store._pool, source_name="faq")
await source.init()
```
Uses PostgreSQL `tsvector` with `ts_rank_cd` for relevance scoring. Auto-creates schema, supports room-scoped queries, and upserts on conflict.
### Response Scoring
Score AI responses automatically using the `ScoringHook`:
```python
from roomkit.scoring import ScoringHook, ConversationScorer, Score
class QualityScorer(ConversationScorer):
async def score(self, *, response_content, query, room_id, channel_id, **kwargs):
# Your scoring logic (LLM-as-judge, rules, heuristics)
return [Score(value=0.9, dimension="relevance", reason="On topic")]
hook = ScoringHook(scorers=[QualityScorer()])
hook.attach(kit)
# Scores are stored as Observations and accessible via hook.recent_scores
```
### User Feedback
Collect user quality ratings:
```python
await kit.submit_feedback("room-1", rating=0.9, comment="Very helpful", dimension="helpfulness")
# Stored as Observation in ConversationStore, fires ON_FEEDBACK hook
```
## Tool Call Events
AIChannel automatically broadcasts ephemeral `TOOL_CALL_DELTA`,
`TOOL_CALL_START`, and `TOOL_CALL_END` events when executing tools. Subscribe
to these for UI indicators:
```python
await kit.subscribe_room("room-1", my_callback)
# Callback receives:
# TOOL_CALL_DELTA: {tool_calls: [{id, name, arguments_chars}], round, channel_id}
# TOOL_CALL_START: {tool_calls: [{id, name, arguments}], round, channel_id}
# TOOL_CALL_END: {tool_calls: [{id, name, result}], round, channel_id, duration_ms}
```
`TOOL_CALL_DELTA` is streaming-only. It reports the running argument size while
the model composes a call, but never exposes the argument content. An empty
`tool_calls` is the terminal frame for that composition, including cancellation,
provider failure, retry, and fallback. Each retry is a new attempt, so its
cumulative `arguments_chars` counts restart from zero. These projection events
are not persisted; the complete arguments still arrive in `TOOL_CALL_START`.
---
`OllamaAIProvider` talks to the native Ollama `/api/chat` endpoint
through the official `ollama-python` SDK. Use it when you want
features the OpenAI-compatible shim hides — chiefly the `think`
parameter and streamed `thinking` deltas.
## Install
```bash
pip install roomkit[ollama]
```
## When to use this vs OpenAI-compat
Ollama also exposes an OpenAI-compatible endpoint at
`http://host:11434/v1/chat/completions`, which works fine with
`OpenAIAIProvider` (or `create_vllm_provider`). Pick `OllamaAIProvider`
when **any** of these matter:
| You want | OpenAI-compat | `OllamaAIProvider` |
|---|---|---|
| Disable reasoning on a thinking model (`think=False`) | Silently ignored | Honored |
| Force reasoning on a non-default model (`think=True`) | Silently ignored | Honored |
| Stream reasoning tokens to a UI as they arrive | Returned in a non-streamed `reasoning` field that the SDK consumer doesn't split out | Streamed as `StreamThinkingDelta` events token-by-token |
| Pass `keep_alive`, `num_ctx`, or `num_predict` cleanly | Awkward via `extra_body` | First-class config |
For plain non-reasoning models (`llama3.2`, `gemma2`, etc.) either
provider works.
## Quick start
```python
from roomkit.providers.ollama import OllamaAIProvider, OllamaConfig
provider = OllamaAIProvider(OllamaConfig(
host="http://localhost:11434",
model="qwen3:8b",
))
# Use exactly like any other AIProvider — pass it to AIChannel, etc.
```
## Config knobs
```python
OllamaConfig(
host="http://localhost:11434", # Ollama server
model="qwen3:8b", # any pulled model
max_tokens=None, # → options.num_predict
temperature=0.7, # → options.temperature
timeout=120.0, # long: cold-start + reasoning is slow
think=None, # None = model default, True/False = explicit
keep_alive="5m", # how long the model stays loaded
num_ctx=8192, # context window override
)
```
## How `think` is resolved
Per-request precedence (highest first):
1. **`AIContext.thinking_budget`** — if set, this wins:
- `None` or `0` → `think=False`
- any `>0` → `think=True`
2. **`OllamaConfig.think`** — fallback when `thinking_budget` is unset.
3. **Omit `think`** — let the model decide its default (reasoning
models think, others don't).
This makes the "Leave empty to disable" semantics in higher-level
agent configs actually honor themselves: a missing/zero
`thinking_budget` at the agent layer reaches the provider as
`think=False` and the model genuinely skips the reasoning phase
instead of just having its reasoning silently discarded.
## Streamed events
`generate_structured_stream()` yields:
- `StreamThinkingDelta(thinking="...")` — one per `message.thinking`
delta chunk from Ollama. Arrives **token-by-token**, which is the
whole reason this provider exists.
- `StreamTextDelta(text="...")` — one per `message.content` delta.
- `StreamToolCall(id, name, arguments)` — collected from
`message.tool_calls` and yielded after the text/thinking deltas
for the same chunk (Ollama doesn't fragment tool-call arguments
across chunks the way OpenAI does).
- `StreamDone(finish_reason, usage)` — terminator.
## History round-trip
The provider preserves `AIThinkingPart` in the message history by
sending it back to Ollama as a top-level `thinking` field on the
assistant message — no `...` tag wrapping needed.
This keeps reasoning models honest across tool-loop rounds: they
see their own prior chain-of-thought when computing the next turn.
## Tool calls
Ollama's tool-call format is essentially OpenAI's, so the standard
`AITool` definitions pass through unchanged. The one quirk: Ollama
doesn't issue stable `id` fields on tool calls, so the provider
synthesizes ones like `call__` so consumers can pair
calls with their results.
## Interactive test bed
`examples/ollama_cli.py` exercises every knob in one place — `--think`,
`--no-think`, `--stream`, `--no-stream`, `--mcp ` for MCP tool
discovery — with a Rich-powered display that labels thinking output
in italics and answer output in bold.
```bash
# Default: model decides whether to think, streams tokens, no tools
uv run python examples/ollama_cli.py --model qwen3:8b
# Force thinking off — fast, single-pass response
uv run python examples/ollama_cli.py --model qwen3:8b --no-think
# Wire in MCP tools
uv run python examples/ollama_cli.py --model qwen3:8b \
--mcp http://localhost:8080/mcp
```
At the prompt, `/think on|off` and `/stream on|off` toggle the
behavior mid-session; `/tools` lists what's available; `/reset`
clears history; `/quit` exits.
## Errors
`ollama.ResponseError` maps to `ProviderError(retryable=...)` where
`retryable=True` for `429`, `500`, `502`, `503`. Transport/connection
errors (timeouts, refused connections) get marked retryable so the
calling `RetryPolicy` decides whether to act.
## What this provider doesn't do
- **Image input**: `supports_vision` reports `True` and image parts
reach the wire, but it's the server-side model that decides whether
to honor them. Unsupported models 400 the request — RoomKit
surfaces that as a non-retryable `ProviderError`.
- **Embeddings**: out of scope. Use `ollama.AsyncClient().embeddings()`
directly if you need them.
---
`ACPChannel` makes RoomKit an **Agent Client Protocol client**: an external coding agent (Claude Code, Codex CLI, Gemini CLI, any registry-listed ACP agent) speaks ACP v1 — spawned here as a subprocess over stdio by default, or reached through a caller-supplied `ACPTransport`; each Room becomes one session of that agent. The reverse direction — exposing a RoomKit agent as an ACP *server* — is out of scope. `CLIChannel` is the interactive terminal transport playing the human side in local sessions.
```bash
pip install "roomkit[acp]" # agent-client-protocol>=0.11.0,<0.12
pip install "roomkit[console]" # rich>=13.0, for CLIChannel(markdown=True)
```
The SDK is imported lazily: `import roomkit` works without the extra; the first connection raises an actionable `ImportError`. RoomKit pins stable ACP wire protocol v1 and rejects any other negotiated `protocolVersion`.
## ACPChannel
```python
from roomkit import ACPChannel
agent = ACPChannel(
"coding-agent",
command=["npx", "-y", "@agentclientprotocol/claude-agent-acp@0.61.0"],
transport=None, # or an ACPTransport, instead of command
cwd="/srv/workspaces/my-project", # required, absolute — on the AGENT's host
additional_directories=None, # extra absolute dirs for the session
env=None, # added to the SDK's restricted env
mcp_servers=None, # ACP MCP-server descriptors (SDK types)
authentication_method=None, # optional ACP auth method id
external_tool_handler=None, # permission policy; None = reject all
)
```
`command` is an argument vector executed directly, **no shell**; a bare string, empty/non-string args, or non-absolute `cwd`/`additional_directories` raise `ValueError`. Exactly one of `command` / `transport` is required (neither or both → `ValueError`), and `env`/`inherit_env` next to a `transport` raise rather than being ignored — they configure the spawn. Class attrs: `channel_type = ChannelType.AI`, `category = ChannelCategory.INTELLIGENCE`, `direction = BIDIRECTIONAL`; capabilities TEXT + RICH. `handle_inbound()` raises `NotImplementedError` — the channel reacts to Room events via `on_event()`.
### Transports
`ACPTransport` (ABC, `channels/acp_transport.py`) is the pipe, and nothing more: `open(client, *, queue) -> ClientSideConnection` (build it with `acp.connect_to_agent(client, writer, reader, queue=queue)` — the protocol only needs a reader/writer pair), `close()` (must not raise; called on teardown *and* on a failed handshake), `is_alive()` (default `True`), and a `name` property surfaced as `info["transport"]`. `StdioACPTransport(command, cwd=…, env=…, inherit_env=…)` is the default, constructed for you from `command=`, and owns the spawn: argument-vector validation, `_resolve_spawn_env`, the stderr drain, and `returncode`-based liveness. Everything protocol-level — `initialize`, version negotiation, `authenticate`, sessions, prompts, permissions, config options, event mapping — stays on the channel, so a transport inherits it.
### Process and session model
One connection per channel, opened lazily on the first prompt; one ACP **session** per Room, created on demand and tagged with extension key `roomkit.live/roomId`. Prompts are serialized per Room (per-room lock); different Rooms progress concurrently through the same connection. When the transport reports the connection dead (for stdio: the subprocess exited), the next prompt reconnects and clears all session mappings — a reconnect never resumes sessions. The client declares no fs/terminal capabilities: `fs/*`, `terminal/*`, and `session/request_input` (elicitation) requests get `method_not_found` — `ON_USER_INPUT_REQUIRED` never fires from ACP.
Methods: `session_id(room_id)` returns the process-local session id or `None`; `cancel(room_id)` cancels the active turn (`True` if a cancel was sent); `close_session(room_id)` closes and forgets one Room's session; `close()` cancels turns, closes sessions, closes the transport and stops the handler. The `info` property reports `{transport, protocol_version, sdk_version, connected, agent, session_count}`.
### How agent output enters the room
For each non-self text event, `on_event()` returns `ChannelOutput(responded=True, response_stream=...)` (TOOL_CALL_START/END events are skipped). The prompt (tagged `roomkit.live/eventId`) yields a `StreamDelta` stream consumed by the inbound-streaming pipeline:
- `agent_message_chunk` → `str` deltas → streamed to transports, persisted as the response event.
- `agent_thought_chunk` → `ThinkingDeltaMarker` in the stream, plus ephemeral `THINKING_START` / `THINKING_DELTA` (thinking truncated to 1000 chars) / `THINKING_END`.
- `tool_call` / `tool_call_update` → `ToolCallStartMarker` / `ToolCallEndMarker` in the stream (persisted as `TOOL_CALL_START`/`TOOL_CALL_END` RoomEvents) plus matching ephemeral events (`result` truncated to 500 chars, `duration_ms`); non-terminal progress → ephemeral `CUSTOM` `{"type": "acp_tool_progress"}`.
- `plan` / `plan_update` / `plan_removed` → ephemeral `CUSTOM` `{"type": "acp_plan_update", session_id, update}`. **Not** `ON_PLAN_UPDATED` — that hook belongs to AIChannel's `plan_tasks` tool.
- `usage_update` → ephemeral `CUSTOM` `{"type": "acp_usage"}`.
`register_channel()` wires `kit`'s realtime backend in automatically. A failed prompt surfaces as `ProviderError(provider="acp")`; cancellation ends the stream silently and closes any open thinking block.
Every output starts a live response record containing
`{"acp": {"protocol_version": "..."}}`. When the prompt returns for a reason
other than `end_turn`, the channel adds `stop_reason`; if it never returns
because of an exception or cancellation, it adds `interrupted: true`. A clean
turn adds neither marker. The final record is available on
`InboundResult.response_metadata`, even when the last activity was a tool call
and no final `MESSAGE` was persisted. With `defer_delivery=True`, await
`result.delivery.wait()` before reading it.
A turn never outlives its tool calls. Whichever way it ends — error, cancellation, or a stop the user asked for, which returns through the ordinary end of a prompt — every tool started without a terminal `tool_call_update` is closed first: a `ToolCallEndMarker` with `status="failed"` and an `error` saying the turn ended before the tool reported, emitted into the stream so the stored `TOOL_CALL_END` exists, plus the matching ephemeral. A turn whose tools all reported emits nothing extra. One gap remains by construction: a stream closed from the outside (its consumer cancelled, a muted binding) is past yielding, so only the ephemeral goes out and the stored row stays pending.
ACP fixes the envelope and leaves the payload to the agent, so `CLIChannel(console=True)` unwraps rather than prints (`roomkit.console._tool_preview`): ACP `text`/`diff` blocks, MCP `content`, and `raw_output` wrappers (`formatted_output`+`exit_code` from Codex, `output`, `result`/`error`) all reduce to their text; a `terminal` block carries no text, so the preview falls back to `raw_output`; `image`/`audio`/`resource` blocks are named, never dumped as base64; 5 lines per result, 200 chars per line; unknown shapes render as compact JSON.
### Permission flow
Every agent `session/request_permission` goes to the `external_tool_handler` (`ExternalToolHandler` ABC from `roomkit.tools`, with `PolicyExternalToolHandler` and `ToolDecision`):
1. `process_tool_call(tool_name, tool_input, *, tool_call_id, session_id, room_id, ...)` → `ToolDecision(approved=...)`. Calling `self._fire_before_hook(...)` fires **`BEFORE_TOOL_USE`** sync hooks (callbacks injected at `register_channel`).
2. Approved → RoomKit selects the agent's `allow_once`/`allow_always` option; denied → `reject_once`/`reject_always`; no matching option → `DeniedOutcome(outcome="cancelled")`.
3. **No handler ⇒ every permission request is rejected.** `ToolDecision.modified_input`/`result` overrides cannot be applied over ACP — setting them logs a warning and rejects the call.
4. On tool completion, `on_tool_result(...)` runs; via `_fire_on_tool_hook` it fires **`ON_TOOL_CALL`** hooks.
## CLIChannel
Interactive terminal transport (`channel_type = ChannelType.CLI`, TEXT only): reads stdin, prints agent output to stdout with ANSI colors.
```python
cli = CLIChannel(
"cli", # channel_id, default "cli"
prompt="You: ",
user_color="\033[33m", # yellow
agent_color="\033[36m", # cyan
thinking_color="\033[2;3m", # dim italic
use_color=None, # None = auto-detect TTY
agent_label=None, # channel_id -> display name ("agent-researcher" -> "Researcher")
show_thinking=False, # render ThinkingDeltaMarker chunks
markdown=False, # live Markdown rendering, requires roomkit[console]
)
```
- `sender_is_participant = True`: `run()`'s `sender_id` is a room **Participant ID**, not an address — identity resolution is skipped for typed lines.
- `deliver()` skips self-echo and prints `Label: text`. `supports_streaming_delivery` is `True`; `deliver_stream()` renders text deltas as they arrive, thinking above the answer, tool events inline (`🔧 name {args}`, `✓`/`✗ name (N ms)`).
- `run(kit, room_id, *, sender_id="user", welcome=None, content_factory=None)` — input loop in a worker thread; each line becomes `kit.process_inbound(...)` with `TextContent(body=line)`, or `content_factory(line)` (`None` skips the line). `quit`/`exit`/`q` or Ctrl+D exits.
## Example: terminal → Claude Code
Condensed from `examples/acp_claude_code.py` (requires `roomkit[acp,console]`, Node.js 22+):
```python
from roomkit import ACPChannel, ChannelCategory, CLIChannel, RoomKit
from roomkit.tools import ExternalToolHandler, ToolDecision
class TerminalPermissionHandler(ExternalToolHandler):
async def process_tool_call(self, tool_name, tool_input, *, tool_call_id="",
room_id=None, **kwargs) -> ToolDecision:
if not await self._fire_before_hook(tool_name, tool_input,
tool_call_id=tool_call_id, room_id=room_id):
return ToolDecision(approved=False, reason="Denied by BEFORE_TOOL_USE hook")
answer = await asyncio.to_thread(input, f"Allow {tool_name}? [y/N] ")
return ToolDecision(approved=answer.strip().casefold() in {"y", "yes"})
async def on_tool_result(self, tool_name, tool_input, result, **kwargs) -> None:
await self._fire_on_tool_hook(tool_name, tool_input, result) # ON_TOOL_CALL hooks
kit = RoomKit()
cli = CLIChannel("you", show_thinking=True, markdown=True,
agent_label=lambda _cid: "Claude Code")
claude = ACPChannel(
"claude-code",
command=["npx", "-y", "@agentclientprotocol/claude-agent-acp@0.61.0"],
cwd=workspace, # absolute Path to the project
env={"ANTHROPIC_API_KEY": api_key, "MAX_THINKING_TOKENS": "1024"},
external_tool_handler=TerminalPermissionHandler(),
)
kit.register_channel(cli)
kit.register_channel(claude)
await kit.create_room(room_id="claude-code-cli")
await kit.attach_channel("claude-code-cli", "you")
await kit.attach_channel("claude-code-cli", "claude-code",
category=ChannelCategory.INTELLIGENCE)
await cli.run(kit, room_id="claude-code-cli") # blocks until quit/Ctrl+D
await kit.close()
```
Flow: terminal line → `CLIChannel` inbound → Room broadcast → `ACPChannel.on_event` → ACP prompt to Claude Code; deltas, thinking, and tool activity stream back through the Room and render live in the terminal, each permission approved once at the prompt.
---
`ImageProvider` (RFC §25) is the surface for an agent that draws. It is
deliberately **separate from the AI provider**, the way STT and TTS already
are: the model holding a conversation is rarely one that draws, so an agent
conversing through Anthropic draws with a Gemini, OpenAI, or xAI key. Nothing
about `AIResponse` changes — images never ride the conversational response.
```python
from roomkit.providers.openai import OpenAIImageConfig, OpenAIImageProvider
images = OpenAIImageProvider(
OpenAIImageConfig(api_key="sk-...", model="gpt-image-2", quality="high")
)
results = await images.generate("an origami fox", size="1024x1024", n=1)
results[0].data # "data:image/png;base64,..." — always a data URI
results[0].mime_type # "image/png"
results[0].decoded() # raw bytes
results[0].to_image_part() # AIImagePart — AI input, or the next edit's reference
await images.close()
```
`generate()` returns **exactly `n` results or raises** — never fewer without
an error. A size the model cannot produce raises rather than silently becoming
another geometry.
## Providers
| Provider | Import | Extra | Notes |
|---|---|---|---|
| `OpenAIImageProvider` | `roomkit.providers.openai` | `roomkit[openai]` | `/v1/images`; `gpt-image-2` takes near-arbitrary sizes, older models a fixed menu — the vendor judges |
| `GeminiImageProvider` | `roomkit.providers.gemini` | `roomkit[gemini]` | Interactions API; size translated to aspect ratio + tier (`512`/`1K`/`2K`/`4K`) |
| `XAIImageProvider` | `roomkit.providers.xai` | `roomkit[xai]` | Grok Imagine; size translated to ratio + `1k`/`2k`; edits go as JSON to `/v1/images/edits` |
| `OpenRouterImageProvider` | `roomkit.providers.openrouter` | `roomkit[openrouter]` | OpenRouter's Image API — 40+ aggregated models (Seedream, FLUX, Recraft, ...); billed cost surfaces as `usage["cost"]` |
| `AzureImageProvider` | `roomkit.providers.azure` | `roomkit[azure]` | Azure OpenAI deployment; `model` is the deployment name |
| `MockImageProvider` | `roomkit.providers.image` | none | Draws a real 1×1 PNG; records `calls`; whole path testable offline |
Configs are per-vendor (`OpenAIImageConfig`, `GeminiImageConfig`,
`XAIImageConfig`, `OpenRouterImageConfig`, `AzureImageConfig`) and carry the
vendor knobs — OpenAI's `quality`/`background`/`output_format`, Gemini's
`image_size`/`output_mime_type`, xAI's `quality`/`resolution`, Azure's
`azure_endpoint`/`api_version`. The `generate()` call is identical everywhere.
```python
from roomkit.providers.xai import XAIImageConfig, XAIImageProvider
images = XAIImageProvider(XAIImageConfig(api_key="xai-...")) # grok-imagine-image-2.0
from roomkit.providers.openrouter import OpenRouterImageConfig, OpenRouterImageProvider
images = OpenRouterImageProvider(
OpenRouterImageConfig(api_key="sk-or-...", model="bytedance-seed/seedream-5-0-pro")
)
from roomkit.providers.azure import AzureImageConfig, AzureImageProvider
images = AzureImageProvider(
AzureImageConfig(
api_key="...",
azure_endpoint="https://myresource.openai.azure.com",
model="my-gpt-image-deployment",
)
)
```
## Editing
Editing is `generate()` with `reference_images`, not a second method:
```python
[original] = await images.generate("an origami fox")
[edited] = await images.generate(
"make the paper blue",
reference_images=[original.to_image_part()],
)
```
Each provider absorbs its vendor's split (OpenAI: multipart `images.edit`;
Gemini: same-call content; xAI: JSON `/images/edits`; OpenRouter:
`input_references` on the same request). A provider that cannot edit reports
`supports_editing == False` and **raises** on references rather than quietly
generating from the prompt alone. `XAIImageProvider` answers per model:
`grok-imagine-image-2.0` and `-quality` edit, the base `grok-imagine-image`
does not.
## The data-URI invariant
`ImageResult.data` is always `data:;base64,` — never bare
base64, never a remote URL. `MediaContent.url` accepts data URIs, so a
generated image enters a room without conversion:
```python
from roomkit.models.event import MediaContent
await kit.send_event(
room_id=room_id,
channel_id="bot",
content=MediaContent(url=result.data, mime_type=result.mime_type),
addressed_to=[], # the picture answers the turn; unaddressed, it would
# re-enter as a fresh prompt and the agent would draw again
)
```
The usual wiring is a tool: the `AIChannel` calls a `generate_image` tool, the
handler draws through the `ImageProvider` and posts the `MediaContent`, and
returns *text* to the model ("Drew 1 image"). See
`examples/image_generation.py` for the complete runnable version.
## Usage and cost
`ImageResult.usage` carries up to four **disjoint** counters — `input_tokens`,
`input_image_tokens`, `output_tokens`, `output_image_tokens` — each token
counted exactly once; counters a vendor does not report are absent, not zero.
- OpenAI and Gemini meter per token: their catalog entries carry
`ModelPricing` with `image_input_per_million` / `image_output_per_million`,
and `entry.pricing.cost_for(result.usage)` prices a generation in USD.
- xAI and most of OpenRouter's lineup bill a flat amount per image: their
catalog entries deliberately carry **no** `pricing` (a per-image charge
restated per token would be wrong, not missing). OpenRouter reports what it
billed on every call, surfaced as `result.usage["cost"]` (USD).
## Catalogs
`available_models()` (classmethod, offline) lists each provider's image
models, tagged `capabilities=["image_gen", "edit"]`. The image catalog is
**disjoint** from `AIProvider.available_models()` — no id draws *and*
converses. `catalog_entry()` returns the active model's `ModelInfo` or `None`.
Azure returns an empty catalog (deployments are user-named); OpenRouter's is a
curated slice — the live set is OpenRouter's public
`GET /api/v1/images/models`.
## Testing
```python
from roomkit.providers.image import MockImageProvider
images = MockImageProvider() # or MockImageProvider(images=[png_bytes])
[result] = await images.generate("anything")
assert result.decoded().startswith(b"\x89PNG")
assert images.calls == [("anything", None, 1, [])]
MockImageProvider(supports_editing=False) # raises on reference_images
```
---
VoiceChannel handles real-time audio conversations with speech-to-text, text-to-speech, and an audio processing pipeline.
## Basic Voice Setup
```python
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(transcripts=["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
```python
# 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:
```python
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",
))
```
### STT language per session
A streaming STT fixes its language when the stream opens. RoomKit opens one
stream per utterance (VAD mode) or per turn (continuous mode), so the
language can change between them: `VoiceChannel.set_stt_language(session,
"fr-CA")` applies from the session's next stream, `None` returns to the
provider's configuration, `get_stt_language(session)` reads it back. Only
providers with `supports_language_override` (Deepgram, the mock) are handed
a language. `TranscriptionResult.language`, `TranscriptionEvent.language`
and `PartialTranscriptionEvent.language` carry what the provider reported —
Deepgram Nova-3 in `language="multi"` reports the language most words were
tagged with; a pinned stream reports nothing.
`STTLanguageLock` packages the detect-then-lock loop:
```python
from roomkit import STTLanguageLock, VoiceChannel
voice = VoiceChannel(
"voice",
stt=DeepgramSTTProvider(DeepgramConfig(api_key="...", model="nova-3", language="multi")),
backend=backend,
pipeline=AudioPipelineConfig(),
stt_language_lock=STTLanguageLock(
detect_language="multi", # every session starts here
prefer={"fr": "fr-CA"}, # reported -> locked
lock_after=1, # agreeing finals before locking
release_after=2, # consecutive misses (no text, low confidence,
), # another language) before detecting again
)
```
Or from a hook, with the same primitives:
```python
@kit.hook(HookTrigger.ON_TRANSCRIPTION)
async def pin_language(event, ctx):
if event.language == "fr":
voice.set_stt_language(event.session, "fr-CA")
return HookResult.allow()
```
## 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 |
| Gemini TTS | `GeminiTTSProvider` | `GeminiTTSConfig` | `roomkit[gemini]` |
| Mock | `MockTTSProvider` | — | built-in |
```python
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 |
```python
from roomkit.voice import get_local_audio_backend
LocalAudioBackend = get_local_audio_backend()
backend = LocalAudioBackend(sample_rate=16000, channels=1)
```
## Shared Capture Sources
A backend takes the microphone when a session starts and releases it when the
session ends. An `AudioCaptureSource` owns the device instead, so anything that
must listen *before* a session exists — a wake word, a level meter — keeps
running, and the session becomes one subscriber among several.
| Source | Class | Extra | Use Case |
|--------|-------|-------|----------|
| Local microphone | `LocalMicSource` | `roomkit[local-audio]` | Wake word, always-on capture |
| Mock | `MockCaptureSource` | built-in | Testing |
```python
from roomkit.voice.capture import LocalMicSource
from roomkit.voice.backends.local import LocalAudioBackend
mic = LocalMicSource(sample_rate=24000, backlog_seconds=10)
mic.start() # the device opens once
detector = mic.subscribe(enqueue, name="wakeword") # no session in sight
transport = LocalAudioBackend(source=mic) # a subscriber, not the owner
mark = mic.mark() # at SPEECH_START
await channel.start_session( # once the trigger matched
room_id, participant_id, connection=None,
metadata={"capture_since": mark},
)
```
The source retains recent audio in a bounded ring. `mark()` names a position in
it and `subscribe(since=mark)` replays from there before going live; a mark that
has aged out replays what remains and reports `truncated`. Replayed frames take
the ordinary inbound path — no new control point.
Fan-out is synchronous on the capture thread, which keeps AEC's capture and
reference timing in step, so a subscriber must enqueue the frame and return.
`start()` / `stop()` alone acquire and release the device: dropping to zero
subscribers never stops capture, and a detector can safely detach for the
duration of a call — which it should, since source frames are pre-AEC.
`LocalAudioBackend(source=...)` derives its input format from the source and
rejects a contradicting explicit argument; with no source its behaviour is
unchanged.
## Interruption Handling
Four strategies for handling user speech during TTS playback:
```python
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:
```python
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:
```python
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
```python
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:
```python
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:
```python
# 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
```python
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:
```python
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).
```python
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.
```python
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.
```python
# 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()`.
Reference audio is converted to the exact post-resampler capture format, and
AEC state and playback activation are isolated per voice session.
While playback is active, realtime backends keep the render timeline aligned
with capture by feeding hardware silence during jitter gaps. Ending playback
bypasses AEC but preserves its converged filter until the session ends.
### AGC (Automatic Gain Control)
Normalizes audio volume levels. AGC does not remove background noise; it makes
quiet and loud speech reach downstream denoising, VAD, and STT at a predictable
level.
```python
from roomkit.voice.pipeline import AGCConfig, AudioPipelineConfig, SimpleAGCProvider
agc = SimpleAGCProvider(AGCConfig(target_level_dbfs=-12.0))
# Or let the pipeline create the built-in provider from the config:
pipeline = AudioPipelineConfig(agc_config=AGCConfig(target_level_dbfs=-12.0))
```
Gain state is isolated per stream, near-silence is not amplified, and a peak
limiter prevents PCM clipping. Processed frames expose
`metadata.gain_applied_db`. Backends declaring `NATIVE_AGC` skip this stage.
### Denoiser
Removes background noise from audio.
```python
# Continuous WebRTC NS (same optional dependency as WebRTC AEC)
from roomkit.voice.pipeline import WebRTCNoiseSuppressorProvider
denoiser = WebRTCNoiseSuppressorProvider(sample_rate=24000)
# 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())
```
WebRTC NS, RNNoise, and ai|coustics accept arbitrary caller chunk sizes without
duplicating or dropping audio; an irregular stream uses a fixed one-native-block
delay (10 ms for WebRTC NS and RNNoise). All built-in denoisers validate PCM
format before creating native state and keep model history isolated per stream.
### Diarization
Identifies different speakers in multi-speaker audio.
```python
from roomkit.voice.pipeline.diarization.mock import MockDiarizationProvider
diarizer = MockDiarizationProvider()
```
### DTMF Detection
Detects dual-tone multi-frequency signals (phone keypad tones).
```python
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.
```python
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):
```python
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:
```python
from roomkit.voice.pipeline.backchannel.mock import MockBackchannelDetector
bc = MockBackchannelDetector()
```
Used with `InterruptionStrategy.SEMANTIC`.
### Post-Processors
Custom audio transformations on outbound TTS audio:
```python
from roomkit.voice.pipeline.postprocessor.base import AudioPostProcessor
```
## Interruption Handling
The `InterruptionHandler` manages what happens when the user speaks during TTS playback:
```python
from roomkit.voice.interruption import InterruptionConfig, InterruptionStrategy
config = InterruptionConfig(
strategy=InterruptionStrategy.CONFIRMED,
min_speech_ms=300,
allow_during_first_ms=600,
)
```
For `RealtimeVoiceChannel` with provider-side VAD, `allow_during_first_ms`
guards the provider input after physical playback begins. RoomKit continues to
process the real microphone signal through AEC, AGC, denoising, and recording,
but forwards equal-duration PCM silence to the provider until the guard expires.
This gives an acoustic echo canceller time to converge without letting residual
onset echo trigger a server-side interruption. The guard requires a transport
with playback callbacks; set it to `0` on a headset or another already-clean
audio path when immediate barge-in is more important.
| 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`:
```python
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:
```python
from roomkit.voice.pipeline import (
MockVADProvider,
MockDenoiserProvider,
MockDiarizationProvider,
MockAGCProvider,
MockAECProvider,
MockDTMFDetector,
MockAudioRecorder,
MockTurnDetector,
MockBackchannelDetector,
)
```
Example with mock VAD:
```python
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
```python
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 Realtime audio (reasoning-capable) |
| xAI Grok | `XAIRealtimeProvider` | — | Grok speech-to-speech |
| Deepgram Voice Agent | `DeepgramAgentProvider` | `roomkit[realtime-deepgram]` | Nova listen + LLM think + Aura speak, each chosen separately |
| Mock | `MockRealtimeProvider` | built-in | Testing |
```python
# OpenAI Realtime
from roomkit.voice import get_openai_realtime_provider
OpenAIRealtime = get_openai_realtime_provider()
provider = OpenAIRealtime(api_key="sk-...") # defaults to gpt-realtime-2.1
# 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 |
```python
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
```python
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:
```python
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):
```python
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:
```python
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:
```python
# OpenAI Realtime supports session resumption
provider = OpenAIRealtime(
api_key="sk-...",
model="gpt-realtime-2.1",
)
# Sessions automatically resume when possible
```
## Hooks
```python
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}")
```
---
Real-time video channels with pluggable processing pipelines, vision AI, bridging, recording, and avatar output.
## Video Channels
| Channel | ChannelType | Use Case |
|---------|-------------|----------|
| `VideoChannel` | `VIDEO` | Video-only transport (webcam, screen, WebSocket, WebRTC) |
| `AudioVideoChannel` | `AUDIO_VIDEO` | Combined A/V backends (SIP, RTP, FastRTC); extends `VoiceChannel` |
| `RealtimeAudioVideoChannel` | `REALTIME_AUDIO_VIDEO` | Speech-to-speech AI with video (e.g. Anam avatar); extends `RealtimeVoiceChannel` |
```python
from roomkit import RoomKit, VideoChannel
from roomkit.video import VideoPipelineConfig
from roomkit.video.backends.local import LocalVideoBackend
from roomkit.video.vision.mock import MockVisionProvider
kit = RoomKit()
video = VideoChannel(
"video-main",
backend=LocalVideoBackend(device=0, fps=30, width=640, height=480),
pipeline=VideoPipelineConfig(vision=MockVisionProvider()),
vision_interval_ms=2000, # vision analysis throttle
bridge=True, # optional: or VideoBridgeConfig
)
kit.register_channel(video)
@kit.on("video_vision_result")
async def on_vision(event):
print(event.data["description"], event.data.get("labels"))
session = await kit.join("webcam-demo", "video-main", participant_id="local-user")
await video.backend.start_capture(session) # LocalVideoBackend-specific
# ... later
await kit.leave(session)
```
`AudioVideoChannel` adds `video_pipeline=`, `vision=`, `vision_interval_ms=`, `avatar=`, `avatar_encoder=`, `video_bridge=` to the `VoiceChannel` kwargs; its `backend` must implement both `VoiceBackend` and `VideoBackend`. `RealtimeAudioVideoChannel` adds `video_pipeline=`, `vision=`, `vision_interval_ms=`; its provider must implement `RealtimeAudioVideoProvider.on_video()`.
## VideoFrame
Inbound frames are `VideoFrame` dataclasses: `data: bytes`, `codec="h264"`, `width=640`, `height=480`, `timestamp_ms`, `keyframe=False`, `sequence`, `metadata`. Encoded codecs: `h264`, `vp8`, `vp9`, `av1`; raw: `raw_rgb24`, `raw_bgr24`, `raw_yuv420p`, `raw_nv12` (`is_encoded`/`is_raw` properties). Outbound streaming uses `VideoChunk` (encoded only).
## Video Pipeline
Inbound order: `[Decoder] -> [Resizer] -> [Transforms...] -> [Filters...] -> taps/bridge/vision`. All stages optional; decoder, resizer, and vision are singular, transforms and filters are chained lists. Vision runs async on its own throttled schedule; the rest is synchronous per-frame.
```python
from roomkit.video.pipeline.decoder.pyav import PyAVVideoDecoder
from roomkit.video.pipeline.resizer.pyav import PyAVVideoResizer
pipeline = VideoPipelineConfig(
decoder=PyAVVideoDecoder(output_format="rgb24"),
resizer=PyAVVideoResizer(width=640, height=480, keep_aspect=True),
transforms=[], filters=[], vision=vision,
recorder=recorder,
recording_config=VideoRecordingConfig(storage="./recordings", format="mp4",
codec="auto", fps=15.0),
)
```
| Stage | Role | Implementations |
|-------|------|-----------------|
| Decoder (`VideoDecoderProvider`) | Encoded → raw pixels; returns `None` until keyframe | `PyAVVideoDecoder`, mock |
| Resizer (`VideoResizerProvider`) | Downscale raw frames to fit target box | `PyAVVideoResizer`, mock |
| Transform (`VideoTransformProvider`) | Per-frame pixel modification (fast, sync) | `VideoEffectTransform(effect=...)`: grayscale, sepia, invert, blur, cartoon, edges, sketch, pixelate; mock |
| Filter (`VideoFilterProvider`) | Inspect/replace frames via `FilterContext` (latest vision result); emit `FilterEvent`s | `YOLODetectorFilter`, `WatermarkFilter`, `CensorVideoFilter`, `FaceTouchFilter` (MediaPipe), `OverlayFilter`, mocks |
| Vision (`VisionProvider`) | Async frame analysis → `VisionResult` | `OpenAIVisionProvider`, `GeminiVisionProvider`, mock |
| Recorder (`VideoRecorder`) | Tap every frame to file | `PyAVVideoRecorder`, `OpenCVVideoRecorder`, mock |
| Encoder (`VideoEncoderProvider`) | Outbound: raw → H.264 NALs (avatar path) | `PyAVVideoEncoder(width=512, height=512, fps=30, codec="libx264")` |
Key filters: `YOLODetectorFilter(model="yolo26n.pt", confidence=0.5, classes=None, every_n_frames=1, draw_boxes=False)`; `CensorVideoFilter(blocked_labels, replacement="black", grace_frames=0)` blanks frames while vision reports a blocked label; `FaceTouchFilter` emits `face_touch` detections. Overlays (`roomkit.video.pipeline.overlay`): `TextOverlayRenderer`, `ImageOverlayRenderer`, `SubtitleManager`/`subtitle_overlay` for live subtitles.
## Video Backends
`VideoBackend` ABC: `connect()`, `accept()`, `disconnect()`, `send_video()`, `send_video_sync()`, `request_keyframe()`, `set_video_passthrough()`; callbacks `on_video_received`, `on_session_ready`, `on_client_disconnected`; `capabilities` flags: `SIMULCAST`, `SVC`, `SCREEN_SHARE`, `RECORDING`, `BANDWIDTH_ESTIMATION`.
| Backend | Class | Extra |
|---------|-------|-------|
| WebSocket | `WebSocketVideoBackend` | fastapi only |
| WebRTC | `FastRTCVideoBackend` (combined A/V) | `roomkit[fastrtc]` |
| RTP | `RTPVideoBackend` (combined A/V) | `roomkit[rtp]` |
| SIP | `SIPVideoBackend` (combined A/V) | `roomkit[sip]` |
| Webcam | `LocalVideoBackend(device=0, fps=30, width=640, height=480)` | `roomkit[local-video]` |
| Screen | `ScreenCaptureBackend(monitor=1, region=None, fps=5, scale=1.0, diff_threshold=0.0)` | `roomkit[screen-capture]` |
| Mock | `MockVideoBackend` | built-in |
`WebSocketVideoBackend` mounts on FastAPI via `mount_websocket_video(app, backend, path=...)` and auto-creates sessions via `set_session_factory()`. Each backend has a `get_*` lazy loader in `roomkit.video` (e.g. `get_sip_video_backend()`).
## Vision Providers
```python
from roomkit.video import OpenAIVisionConfig, OpenAIVisionProvider
# Defaults target Ollama: base_url="http://localhost:11434/v1", model="qwen3.5"
vision = OpenAIVisionProvider(OpenAIVisionConfig(
api_key="sk-...", base_url="https://api.openai.com/v1",
model="gpt-4o", detail="low", max_tokens=100,
))
# GeminiVisionProvider(GeminiVisionConfig(api_key="AIza...", model="gemini-3.1-flash-lite"))
```
`await provider.analyze_frame(frame, prompt=None)` returns `VisionResult(description, labels, confidence, faces, text, metadata)` — `text` is OCR, `faces` are `FaceDetection` boxes. Results are cached per session (`channel.get_last_vision_result(session_id)`), emitted as `video_vision_result` framework events, and auto-injected into the system prompt of any `AIChannel` in the same room. For realtime voice, `setup_realtime_vision(kit, room_id, voice_channel_id)` injects via `inject_text(silent=True)`. On-demand tools: `DescribeWebcamTool`, `DescribeScreenTool`, `ListWebcamsTool`.
## Video Hook Triggers
| Trigger | Fires When | Execution |
|---------|-----------|-----------|
| `ON_VIDEO_SESSION_STARTED` | Backend signals video path live and session bound | async |
| `ON_VIDEO_SESSION_ENDED` | Session unbound / client disconnected | async |
| `ON_VISION_RESULT` | Vision analysis completed (`VisionEvent`) — can block or modify the description | sync |
| `ON_VIDEO_DETECTION` | Pipeline filter emitted a detection (`VideoDetectionEvent`: `kind`, `labels`, `confidence`, `metadata`) | async |
| `BEFORE_BRIDGE_VIDEO` | Frame about to be bridge-forwarded (`BridgeVideoEvent`); `HookResult.block()` drops it | sync |
| `ON_SCREEN_SHARE_STARTED` / `STOPPED` | ConferenceChannel: `SCREEN_SHARE`-kind track published/unpublished | async |
| `ON_VIDEO_TRACK_ADDED` / `REMOVED` | Reserved — defined in `HookTrigger`, not yet fired by built-in channels | — |
## Video Bridge
`bridge=True` (or `VideoBridgeConfig(enabled=True, max_participants=10, forwarding_strategy="forward", keyframe_interval_s=5.0)`) forwards frames between sessions in the same room for human-to-human video, requesting keyframes (PLI) from sources periodically and when receivers join. `channel.set_bridge_filter(fn)` installs a synchronous per-frame filter `(source_session, frame) -> frame | None` — the fast-path alternative to `BEFORE_BRIDGE_VIDEO`.
## Avatars
`AvatarProvider` (`roomkit.video.avatar`) generates lip-synced video from TTS audio on `AudioVideoChannel`: `MuseTalkAvatarProvider` (local GPU), `WebSocketAvatarProvider` (remote inference), `MockAvatarProvider`. Pair with `avatar_encoder=PyAVVideoEncoder(...)` for RTP/SIP output. Cloud avatar A/V (Anam) uses `RealtimeAudioVideoChannel` instead.
## Optional Extras
`roomkit[video]` (PyAV stages: av, numpy) · `[local-video]` (webcam, effects, OpenCV recorder) · `[screen-capture]` (mss) · `[screen-input]` (pyautogui) · `[yolo]` (ultralytics) · `[mediapipe]` · `[video-overlay]` (Pillow) · `[fastrtc]`/`[rtp]`/`[sip]`/`[anam]` (transports / cloud avatar).
---
ConferenceChannel bridges an external SFU conference into a RoomKit room. The SFU carries all human-to-human media; RoomKit joins as a single bot participant for transcription, AI voice, recording, and speech-to-speech — it never proxies human media (RFC §12.10).
## Constructor
```python
ConferenceChannel(
channel_id,
*,
backend, # ConferenceBackend (required)
stt=None, # STTProvider — per-track transcription
tts=None, # TTSProvider — the bot's voice
realtime=None, # ConferenceRealtimeConfig — speech-to-speech (excludes tts)
pipeline=None, # default: 16 kHz mono contract + EnergyVADProvider
interruption=None, # ConferenceInterruptionConfig
recording=None, recorder=None, # ConferenceRecordingConfig + MediaRecorder, both or neither
bot_identity="roomkit",
bot_grants=None, # explicit ConferenceGrants; default derived via for_bot()
default_grants=None, # what mint_access() grants humans; default ConferenceGrants()
e2ee=False, close_room_on_detach=False,
speak_text_events=False, # off: only AI-channel text events are spoken
close_providers=True,
max_queued_frames=100, # per-track backpressure bound (lane + recording)
identity_address_keys=None, identity_trusts_unasserted_metadata=False,
)
```
Refused at construction (and identically at plug time): `e2ee=True` with stt/recording/realtime (bot receives ciphertext); `ConferenceRecordingMode.EGRESS` (only `FRAMEWORK` is implemented); `tts` + `realtime` together (one bot track, one voice); `realtime.tools` without `tool_handler`; a `pipeline` without a VAD when stt/realtime is set.
Public surface: `mint_access()`, `plug_*/unplug_*()`, `set_bot_grants()`, `may_interrupt(participant_id)`, `active_lanes` (dict `track_id -> ConferenceLane`; `drain()`, `dropped_frames`), `info()`, `close()`.
## How the Bridge Works
The bot joins lazily, only when the channel has a need (stt/tts/recording/realtime) and someone is coming: a `mint_access()`, an arrival, an occupancy probe. With no need configured the channel is pure transport — admission gate and roster, no bot in the meeting.
- **Inbound**: each subscribed AUDIO track gets its own `ConferenceLane` (queue + task + VAD state). The VAD segments utterances; one utterance → one STT call → one `RoomEvent` attributed to the track's `participant_id` (track identity replaces diarization). A full lane drops oldest frames.
- **Outbound**: `deliver()` speaks AI-channel TextContent on the single bot track via TTS (all text events if `speak_text_events=True`); with `realtime`, text is injected into the provider's context instead.
- **Video**: never subscribed — `capabilities()` announces AUDIO only.
```python
from roomkit import ConferenceTranscription, HookResult, HookTrigger
@kit.hook(HookTrigger.ON_TRANSCRIPTION) # sync, runs BEFORE the room; block/modify = redaction
async def on_text(payload: ConferenceTranscription, ctx) -> HookResult:
# payload: track_id, participant_id, room_id, text
return HookResult.allow()
```
## Conference Hooks
SYSTEM lifecycle events; payload in `event.content.data` (always includes `channel_id`).
| Trigger | Fired when | Data |
|---------|-----------|------|
| `ON_CONFERENCE_PARTICIPANT_JOINED` / `_LEFT` | SFU reports arrival/departure | `participant_id` |
| `ON_CONFERENCE_TRACK_PUBLISHED` / `_UNPUBLISHED` | Track publish/unpublish | `track_id`, `participant_id`, `kind` |
| `ON_CONFERENCE_TRACK_MUTED` / `_UNMUTED` | Publisher (un)mutes; muted VIDEO = camera off | `track_id`, `participant_id`, `kind` |
| `ON_SCREEN_SHARE_STARTED` / `_STOPPED` | SCREEN_SHARE track published/unpublished | `track_id`, `participant_id` |
| `ON_ACTIVE_SPEAKER_CHANGED` | Dominant speaker (`ACTIVE_SPEAKER` capability) | `participant_id` |
| `ON_CONNECTION_QUALITY_CHANGED` | Quality report (backend label, not normalized) | `participant_id`, `quality` |
`ON_SPEECH_START`/`ON_SPEECH_END` fire per lane at VAD edges. `ON_BARGE_IN` carries `ConferenceBargeIn(room_id, track_id, participant_id, interrupted_text, audio_position_ms)`.
## ConferenceBackend ABC (`roomkit.conference`)
- Control plane: `ensure_room(room_id, metadata=None, e2ee=False)`, `close_room()`, `mint_access(room_id, participant_id, grants, *, display_name=None) -> ConferenceAccess`, `list_participants()`, `remove_participant()`, `mute_track()`, `unmute_track()` (needs `REMOTE_UNMUTE`).
- Bot session: `join_as_bot(room_id, identity, grants) -> BotSession`, `leave(bot)`, `update_bot_grants(bot, grants)` (needs `BOT_GRANT_UPDATE`, else `ConferenceCapabilityError`), `subscribe_track(bot, track_id)`, `unsubscribe_track()`, `publish_audio(bot, chunk)` (PCM `AudioChunk`; `is_final` ends the utterance), `stop_playback(bot)` (barge-in: discard queued unplayed audio; utterance still ends on `is_final`; no-op for a gone session), `publish_video()` (needs `VIDEO_PUBLISH`), `close()`.
- Callbacks (what drives the channel): `on_participant_joined/left`, `on_track_published/unpublished/muted/unmuted`, `on_track_audio`, `on_track_video`, `on_active_speaker_changed`, `on_connection_quality`, `on_bot_session_ended` (SFU dropped the bot without a `leave()`).
- `ConferenceCapability` flags: `SCREEN_SHARE`, `EGRESS_RECORDING`, `SIP_GATEWAY`, `ACTIVE_SPEAKER`, `CONNECTION_QUALITY`, `VIDEO_PUBLISH`, `REMOTE_UNMUTE`, `BOT_GRANT_UPDATE`, `E2EE`.
## Backends
| Backend | Class | Config | Extra |
|---------|-------|--------|-------|
| LiveKit | `LiveKitConferenceBackend` | `LiveKitConfig` | `roomkit[livekit]` |
| Mock | `MockConferenceBackend` | `capabilities=` kwarg | built-in |
```python
from roomkit.conference.livekit import LiveKitConfig, LiveKitConferenceBackend
backend = LiveKitConferenceBackend(LiveKitConfig(url="wss://my-project.livekit.cloud"))
```
`LiveKitConfig`: `url`/`api_key`/`api_secret` fall back to `LIVEKIT_URL`/`LIVEKIT_API_KEY`/`LIVEKIT_API_SECRET`; `access_ttl=timedelta(minutes=15)`, `audio_sample_rate=48_000`, `audio_channels=1`, `publish_queue_ms=300`, `remote_unmute=False`, `sip_gateway=False`, `room_metadata_key="roomkit"`. Declares `SCREEN_SHARE | ACTIVE_SPEAKER | CONNECTION_QUALITY | BOT_GRANT_UPDATE` (+`REMOTE_UNMUTE`/`SIP_GATEWAY` when configured).
`MockConferenceBackend` scripts SFU events: `simulate_participant_joined/left`, `simulate_track_published(room_id, participant_id, kind=TrackKind.AUDIO)`, `simulate_track_unpublished`, `simulate_audio(track, frame)`, `simulate_track_muted/unmuted`, `simulate_active_speaker`, `simulate_connection_quality`, `simulate_bot_disconnected(bot, reason)`; fault injection `fail(method, exc)` / `delay(method, seconds)`; assertion state `calls`, `published_audio`, `utterances`, `subscriptions`, `playback_stops`.
## Models & Grants
- `ConferenceGrants(publish_audio=True, publish_video=True, publish_screen_share=True, subscribe=True, moderate=False, hidden=False)` — human defaults. `ConferenceGrants.for_bot(speaks=False, listens=True)` = least privilege; `ConferenceGrants.observer()` = subscribe-only + hidden.
- `ConferenceAccess(url, token, expires_at, provider_data)` — opaque client credential; `token` excluded from repr.
- `BotSession(id, room_id, identity, joined_at, metadata)`; `ConferenceTrack(id, room_id, participant_id, kind, muted, metadata)`; `ConferenceParticipant(participant_id, display_name, connected_at, tracks, metadata, asserted_metadata)`; `TrackKind.AUDIO/VIDEO/SCREEN_SHARE`.
- `ConferenceInterruptionConfig(strategy=InterruptionStrategy.IMMEDIATE, scope=ConferenceInterruptionScope.ANY, allowlist=[])` — scope `ANY`/`NONE`/`ALLOWLIST` decides who may barge in.
- `ConferenceRealtimeConfig(provider, system_prompt=None, voice=None, tools=None, tool_handler=None, temperature=None, input_sample_rate=24000, output_sample_rate=24000, server_vad=True, provider_config=None)` — lanes mixed N→1 into one speech-to-speech session per room; the provider speaks on the bot track.
## Hot-Plugging
Effects are in force when the plug returns: occupied conferences joined, published tracks subscribed, bot grants realigned (in place with `BOT_GRANT_UPDATE`, announced re-join otherwise). An occupied slot is refused — swap = unplug, then plug. Unplugging the last need makes the bot leave every conference.
```python
await channel.plug_stt(stt) # optional pipeline=; joins an existing pipeline
await channel.plug_tts(tts)
await channel.plug_recording(ConferenceRecordingConfig(), recorder=recorder)
await channel.plug_realtime(config)
await channel.unplug_stt(); await channel.unplug_tts()
await channel.unplug_recording(); await channel.unplug_realtime()
await channel.set_bot_grants(ConferenceGrants.observer()) # None returns to derivation
```
## Shutdown Semantics (RFC §12.10.4)
`close()` runs exactly one shutdown per channel (`ConferenceShutdownCoordinator`): concurrent callers join the same shielded task and the terminal result is replayed on later calls. Order: close admission → stop playbacks → wait teardowns/joins → disconnect realtime sessions → bots leave every conference → close lanes → finalize recordings → close backend → close providers → settle roster writes. Every backend/provider call holds an operation lease; a resource still leased is **retained**, not closed — it closes in the background when its last lease returns, and the close **fails**: `ConferenceCloseError` aggregates structured `CloseIssue`s (`component`, `operation`, `status` in `FAILED`/`TIMED_OUT`/`ABANDONED`/`RETAINED`, `step`, `detail`) plus any bot session that could not be removed (still reported by `info()`). All waits are bounded — a slow store never holds a bot in a meeting.
`info()` answers RFC §17.7 disclosure per room: `bot_present`, `bot_hidden`, `stt_active`, `realtime_active`, `recording_active`, `recording_dropped_frames`, `active_lanes`, `collecting`, `leave_failed`.
## Minimal Example
```python
import asyncio
from roomkit import MockConferenceBackend, RoomKit
from roomkit.channels.conference import ConferenceChannel
from roomkit.voice.stt.mock import MockSTTProvider
async def main() -> None:
backend = MockConferenceBackend()
channel = ConferenceChannel("conf", backend=backend, stt=MockSTTProvider(transcripts=["Hi."]))
kit = RoomKit()
kit.register_channel(channel)
await kit.create_room("standup")
await kit.attach_channel("standup", "conf")
# Mint SFU credentials for a human client; the mint also starts the lazy bot join
await kit.ensure_participant("standup", "conf", "alice", display_name="Alice")
access = await channel.mint_access("standup", "alice") # hand access.url/token to the client
# With a real backend, SFU events drive these
await backend.simulate_participant_joined("standup", "alice", display_name="Alice")
mic = await backend.simulate_track_published("standup", "alice")
# feed AudioFrames: simulate_audio(mic, frame) — speech, then silence — then
# await channel.active_lanes[mic.id].drain() and read kit.store.list_events("standup")
await kit.close()
asyncio.run(main())
```
Runnable references: `examples/conference_quickstart.py`, `conference_ai_meeting.py`, `conference_realtime_ai.py`, `conference_livekit.py`.
---
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.
```python
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.
```python
from roomkit import Swarm
kit = RoomKit(orchestration=Swarm(agents=[billing, shipping, returns]))
```
### Supervisor
A supervisor agent delegates tasks to worker agents in child rooms:
```python
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.
```python
from roomkit import Loop
kit = RoomKit(orchestration=Loop(
agent=writer_agent,
reviewer=editor_agent,
max_iterations=3,
))
```
## Using Orchestration
```python
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):
```python
# The AI calls this tool automatically:
# handoff_conversation(target="handler", reason="Billing issue", summary="User needs invoice help")
```
The handoff:
1. Updates `ConversationState` in room metadata (phase, active agent, handoff count)
2. Mutes the outgoing agent, unmutes the incoming agent
3. Emits a system event visible to the new agent
4. 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. |
| `active_agent_id` | `str \| None` | `None` | Channel ID of the currently active agent. |
| `previous_agent_id` | `str \| None` | `None` | Agent before the last transition. |
| `handoff_count` | `int` | `0` | Total number of agent handoffs. |
| `phase_started_at` | `datetime` | now | When the current phase started. |
| `phase_history` | `list[PhaseTransition]` | `[]` | Immutable audit trail of all transitions. |
| `context` | `dict[str, Any]` | `{}` | **Arbitrary user data** — store custom key-value pairs across turns. |
Built-in phases: `ConversationPhase.INTAKE`, `QUALIFICATION`, `HANDLING`, `ESCALATION`, `RESOLUTION`, `FOLLOWUP`.
### Reading State
```python
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"}
# 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:
```python
from roomkit.orchestration.state import get_conversation_state, set_conversation_state
room = await kit.get_room("support")
state = get_conversation_state(room)
# Store custom data in context (immutable update)
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
```python
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:
```python
from roomkit.orchestration.state import get_conversation_state, set_conversation_state
room = await kit.get_room("support")
state = get_conversation_state(room)
new_state = state.transition(
to_phase="escalation",
to_agent="supervisor-agent",
reason="Customer requested manager",
metadata={"escalation_priority": "high"},
)
updated_room = set_conversation_state(room, new_state)
await kit.store.update_room(updated_room)
```
### 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 (UTC) |
| `metadata` | `dict[str, Any]` | Arbitrary metadata for this transition |
### Using State in Hooks
```python
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)
updated_state = state.model_copy(update={
"context": {
**state.context,
"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()
```
## Conversation Router (Advanced)
`ConversationRouter` dynamically routes incoming messages to different agents based on conversation state, message content, origin channel, or custom logic.
### RoutingConditions Reference
All conditions are ANDed — every non-None field must match:
| 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 `(event, context, state) -> bool` for arbitrary logic |
### Routing by Phase
```python
from roomkit.orchestration.router import ConversationRouter, RoutingRule, RoutingConditions
router = ConversationRouter(
rules=[
RoutingRule(agent_id="billing-agent", conditions=RoutingConditions(phases={"billing"})),
RoutingRule(agent_id="shipping-agent", conditions=RoutingConditions(phases={"shipping"})),
],
default_agent_id="triage-agent",
)
```
### Routing by Channel Type (Origin)
```python
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:
```python
@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
```python
def is_high_value_customer(event, ctx, state):
return state.context.get("customer_tier") == "premium"
def contains_urgency(event, ctx, state):
if isinstance(event.content, TextContent):
return any(w in event.content.body.lower() for w in ["urgent", "emergency", "asap"])
return False
router = ConversationRouter(
rules=[
RoutingRule(agent_id="senior-agent", conditions=RoutingConditions(custom=is_high_value_customer), priority=-1),
RoutingRule(agent_id="escalation-agent", conditions=RoutingConditions(custom=contains_urgency), priority=0),
],
default_agent_id="general-agent",
supervisor_id="supervisor-agent",
)
```
### Combined Conditions
Combine multiple conditions (all are ANDed):
```python
# 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
```python
# Option 1: Manual hook
kit.hook(HookTrigger.BEFORE_BROADCAST, execution=HookExecution.SYNC, priority=-100)(router.as_hook())
# Option 2: install() — also sets up handoff tools
handler = router.install(kit, agents=[billing_agent, shipping_agent, triage_agent])
```
### Routing Selection Priority
0. **Address** — if the event carries `addressed_to`, it decides; the router returns untouched
1. **Agent affinity** — if `state.active_agent_id` is set and attached, stick with it
2. **Rules** — evaluate in ascending `priority` order; first match wins
3. **Fallback** — return `default_agent_id`
4. **Loop prevention** — events FROM intelligence channels are never routed
## Addressing
Routing rules answer *which agent handles this kind of event*. Addressing answers *which agent am I talking to right now*, per message:
```python
await kit.process_inbound(
InboundMessage(
channel_id="you",
sender_id="user",
content=TextContent(body="review hello.py"),
addressed_to=["codex"], # only this agent is asked to act
)
)
```
| `addressed_to` | Meaning |
|---|---|
| `None` | Unaddressed — every eligible agent acts, or the router decides |
| `["codex"]` | Only `codex` is asked; the others see it and stay silent |
| `[]` | Nobody is asked — a decision, not an absence |
Addressing is **not** visibility: it narrows who is *asked*, never who may *see*. Transport delivery is untouched, so the humans in the room still get the message. The address is stored on the event, so a transcript shows who was asked.
Direct injection addresses the same way — `[]` is what an application needs when it stores a message and triggers the answer itself:
```python
await kit.send_event(
room_id=room_id,
channel_id="system",
content=TextContent(body=body),
addressed_to=[], # stored, and asking nobody
)
```
### Agent Response Policy
An agent's own output solicits the other agents by default (the chaining a pipeline needs, bounded by `max_chain_depth`). In a room of independent agents that is a hazard:
```python
# At creation, kit-wide or per room
kit = RoomKit(agent_response_policy=AgentResponsePolicy.ADDRESSED_ONLY)
await kit.create_room(room_id="dev", agent_response_policy=AgentResponsePolicy.ADDRESSED_ONLY)
# On a live room — the one that just gained a second agent
await kit.attach_channel(room_id, "codex", category=ChannelCategory.INTELLIGENCE)
await kit.set_agent_response_policy(room_id, AgentResponsePolicy.ADDRESSED_ONLY)
```
| Policy | An agent's output solicits |
|---|---|
| `AGENT_CHAIN` | every eligible intelligence channel — the default |
| `ADDRESSED_ONLY` | only the channels it addressed, if any |
A policy change applies to events processed after it; setting the policy a room already holds is a no-op. A binding that is not solicited is skipped before any work is done for it, so a roster can be attached lazily and rehydrated one agent at a time.
RoomKit takes the decision, never the syntax: `@codex`, a `/agent` command or a picker all live in the application, which passes channel ids.
## Conversation Pipeline (Advanced)
Define stages explicitly:
```python
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:
```python
# 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:
```python
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:
```python
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
```python
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
```python
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
```python
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
```python
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:
```python
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 VoiceMeUpSMSProvider # use provider.parse_inbound()
# Or use the universal webhook parser
message = await kit.process_webhook(meta=request_data, channel_id="sms-twilio")
```
## RCS
```python
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
```python
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
```python
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)
```python
from roomkit import WhatsAppChannel
from roomkit.providers.whatsapp.base import WhatsAppProvider
whatsapp = WhatsAppChannel("wa-business", provider=WhatsAppProvider(
access_token="...",
phone_number_id="...",
))
```
### Personal (neonize)
```python
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
```python
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
```python
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]`.
Inbound `external_id` and `idempotency_key` are `:` because
Telegram message ids are chat-local. Malformed nested objects and invalid
coordinates are rejected at the webhook boundary.
- Inbound media (`photo`, `voice`, `audio`, `video_note`, `video`, `document`) parses to a `TextContent` whose body is the caption — empty when there is none, as on a voice note — plus `metadata["file_id"]` and `metadata["media_type"]`, and whichever of `duration`, `mime_type`, `file_name`, `file_size` Telegram sent.
- `parse_telegram_message(msg)` is the layer below: it reads a Telegram `message` into `TelegramMessageParts` (content, metadata, `message_id`, `sender_id`) and attributes nothing. `parse_telegram_webhook` is that function plus the ordinary attribution — the sender is `message.from.id`. Use the lower one when your identity model is not Telegram's (a one-bot-per-user deployment attributes a DM to the bot's owner), so that reading a `file_id` never costs you your identity model.
- Resolving that `file_id` to bytes belongs to the provider, which holds the bot token: `path = await provider.get_file(file_id)` then `data = await provider.download_file(path)`. Both return `None` on failure and log a warning that never carries the URL (every Bot API URL embeds the token). Telegram caps Bot API downloads at 20 MB and refuses larger files at the `getFile` step; `metadata["file_size"]` tells you before you spend the call.
- RoomKit stops at the bytes — transcription and storage are the application's call.
- `parse_telegram_message` also gives `entities` (a caption's markup comes from `caption_entities`), `reply_to_message_id` and `media_group_id`. None of them reach the `InboundMessage` — `parse_telegram_webhook`'s metadata is unchanged.
`TelegramBotProvider` is `TelegramBotAPI` plus the rendering of a `RoomEvent`. The API half is the Bot API surface an application needs around its sends, so it never writes a second HTTP client for the same token:
```python
me = await provider.get_me() # metadata["result"] = the bot object
if not me.success:
... # me.error, me.metadata["description"]
await provider.set_webhook(url, secret=secret, allowed_updates=["message", "callback_query"])
await provider.answer_callback_query(cq.id, "Approved.")
await provider.edit_message_text(chat, msg_id, "Approved", reply_markup={"inline_keyboard": []})
```
Also `get_updates`, `delete_webhook`, `leave_chat`, `send_message` (plain text, no Markdown pass), `send_force_reply` (its `provider_message_id` is what a later reply is matched against), `send_chat_action` and `edit_message_reply_markup`. Every call answers with a `ProviderResult`: `telegram_` / `http_` / `timeout` / a safe transport exception class, and Telegram's own words under `metadata["description"]`. Transport errors never echo the token-bearing request URL.
Update reading is two levels. `parse_telegram_update(payload)` says which form arrived — a `message`, an `edited_message` (same shape, `edited=True`), or a `callback_query` parsed into a `TelegramCallback` (`id`, `data`, `sender_id`, `chat_id`, `message_id`, `message_text`). `callback_data` is posted by whoever pressed the button, so treat what it names as a claim to check.
`mentions_bot(msg, bot_username=..., bot_id=...)` says whether a group message addressed the bot — a reply to the bot, `bot_command`, `mention`, `text_mention`, or the handle as plain text. It gives the fact; whether that group answers only when addressed is your policy. `entity_text(text, entity)` slices by an entity's offsets, which count **UTF-16 code units** — a code-point slice returns the wrong substring as soon as an emoji precedes the mention.
## Discord
Discord has no webhook parser — it is a source + provider pair sharing one persistent gateway connection. `DiscordGatewaySource` owns the `discord.Client` (inbound); `DiscordBotProvider` reuses that client for outbound sends.
```python
from roomkit import DiscordChannel, RoomKit
from roomkit.providers.discord import DiscordBotProvider, DiscordConfig
from roomkit.sources.discord import DiscordGatewaySource
config = DiscordConfig(
bot_token="...", # SecretStr
intents_message_content=True, # privileged intent — enable in the Developer Portal
ignore_bots=True, # drop inbound messages authored by other bots
)
source = DiscordGatewaySource(config, channel_id="discord-main")
provider = DiscordBotProvider(source) # sends through the source's client
kit = RoomKit()
kit.register_channel(DiscordChannel("discord-main", provider=provider))
await kit.attach_source("discord-main", source, auto_restart=True) # connects the gateway
```
Requires `pip install roomkit[discord]` (discord.py). The Message Content intent is privileged: enable it under Bot > Privileged Gateway Intents in the Discord Developer Portal, or every inbound `message.content` arrives empty.
- Recipient key `discord_channel_id` resolves the target Discord channel snowflake at delivery time.
- Capabilities: text + rich + media, `max_length=2000`, threading and reactions. `RichContent` is sent as an embed; `MediaContent` with an http(s) URL rides in the message content (Discord auto-embeds), a `data:` URI is decoded and uploaded as a file.
- Threading: outbound `channel_data.thread_id` (a message snowflake) becomes a reply reference; inbound reply references become `InboundMessage.thread_id`.
- Inbound parsing: `parse_discord_message(message, channel_id, bot_user_id=..., ignore_bots=True)` returns `InboundMessage | None` — the bot's own messages (and other bots, by default) are dropped, so echo hooks never loop. Metadata carries `guild_id`, `channel_id`, `channel_name`, `author_name`, `author_bot`, `message_id`. Override via `DiscordGatewaySource(..., parser=...)`.
- Reactions: `provider.send_reaction(channel_id, message_id, emoji)` outbound; inbound reaction add/remove events reach the source's `on_event` callback as normalized dicts (`action`, `emoji`, `user_id`, `message_id`, `channel_id`) — outside the message pipeline.
- Testing: `MockDiscordProvider` records `sent` and `reactions` without the `discord` dependency. ABC: `DiscordProvider`.
Runnable example: `examples/discord_bot.py`.
## Buzz (Nostr)
Buzz (Block's Nostr-based team workspace) follows the same source + provider pairing: `BuzzRelaySource` owns a `buzzkit.BuzzClient` — NIP-42 authentication plus a realtime channel subscription — and `BuzzProvider` reuses that client for outbound sends over the relay's HTTP bridge, so one Nostr identity serves both directions.
```python
from roomkit import BuzzChannel, RoomKit
from roomkit.providers.buzz import BuzzConfig, BuzzProvider
from roomkit.sources.buzz import BuzzRelaySource
config = BuzzConfig(
relay_url="wss://your-community.communities.buzz.xyz",
private_key="nsec1...", # agent's Nostr secret (nsec or hex) — signs events, authenticates (NIP-42)
ignore_own=True, # drop the agent's own events (echo guard)
auto_join=True, # NIP-29 self-join (kind 9000, role=bot) on connect
announce_presence=True, # kind 20001 "online" on connect + periodic heartbeat
auth_tag=None, # optional NIP-OA owner attestation (buzzkit.compute_auth_tag)
)
source = BuzzRelaySource(config, channel_id="buzz-main", relay_channel_id="")
provider = BuzzProvider(source) # sends through the source's client
kit = RoomKit()
kit.register_channel(BuzzChannel("buzz-main", provider=provider))
await kit.create_room(room_id="buzz-room")
await kit.attach_channel("buzz-room", "buzz-main", metadata={"buzz_channel_id": ""})
await kit.attach_source("buzz-main", source, auto_restart=True) # connects + subscribes
```
Requires `pip install roomkit[buzz]` (installs `buzzkit`, a compiled wheel kept out of the aggregate extras). Hosted Buzz communities are closed relays: the agent's key must be a member first — claim an invite once with `buzzkit`'s `claim_invite`, then copy the channel UUID from the Buzz app.
- Recipient key `buzz_channel_id` resolves the target Buzz relay channel UUID at delivery time. Each source subscribes to a single relay channel — register one source per Buzz channel and bind each to its room.
- Capabilities: text only, `max_length=65536`, threading and reactions advertised.
- Inbound parsing: `parse_buzz_event(event, channel_id, own_pubkey=..., ignore_own=True)` converts a kind-9 Nostr event dict to an `InboundMessage` — sender pubkey becomes `sender_id`, the Nostr event id becomes `external_id` and `idempotency_key`, metadata carries `nostr_event_id`, `nostr_kind`, `buzz_channel_id`. Subscribe to other event kinds with `BuzzRelaySource(..., kinds=[...], parser=...)`.
- `provider.send(event, to=channel_uuid)` publishes a kind-9 channel message signed with the agent's key; the returned `ProviderResult.provider_message_id` is the Nostr event id. HTTP-bridge sends succeed even while the inbound WebSocket is mid-reconnect; the source reconnects with exponential backoff (1 s doubling to a 30 s cap).
- Testing: `MockBuzzProvider` records `sent` without the `buzzkit` dependency. ABC: `BuzzRelayProvider`.
- Presence: with `announce_presence=True` the source publishes kind-20001 `"online"` on connect, heartbeats every 30 s (surviving transient publish failures), and publishes `"offline"` on a deliberate `stop()` so the agent's dot flips immediately instead of lapsing by relay TTL.
- Owner control commands (`buzzkit>=0.3.0`): with `obey_owner_commands=True` (default), a kind-9 message whose trimmed content is exactly `!shutdown` / `!cancel` / `!rotate`, mentioning the agent and authored by the **proven** owner — the NIP-OA auth tag's Schnorr-verified attester, else `BuzzConfig.owner_pubkey` — is consumed before the pipeline (the AI never answers its own stop command). `!shutdown` stops the source gracefully; all commands reach the optional `on_owner_command` callback, which takes over the response when provided. No provable owner, or a non-owner author → the message flows normally (fail-closed). Replay-safe: a command issued before the source started is stale (relays replay recent history on every subscribe) — consumed without action; one issued during a disconnection is honored when the reconnect replays it.
- Inbound metadata carries `nostr_created_at` (unix seconds, the Nostr timestamp) so apps can tell live traffic from relay-history replay — used by `examples/buzz_agent.py`'s echo guard.
Runnable example: `examples/buzz_bot.py`.
### BuzzAgent — the lifecycle runner
`BuzzAgent` (`roomkit.providers.buzz`) turns a configured RoomKit app into a first-class Buzz agent: it attaches the sources (taking over their `on_owner_command`), installs SIGTERM/SIGINT handlers, optionally arms an `exit_after_inactivity` bound (seconds; default off; reaper on its own timer), and exits every stop cause — owner `!shutdown`, signal, inactivity — through one graceful path: `kit.close()` (presence `offline`, sockets closed). `run()` is single-shot, consumes the kit, and returns a `BuzzAgentStopCause` (`owner_shutdown` / `signal` / `inactivity`); exit the process with code 0 so supervisors never restart an intentional stop.
```python
from roomkit.providers.buzz import BuzzAgent, BuzzConfig
config = BuzzConfig.from_env() # BUZZ_PRIVATE_KEY (or NOSTR_PRIVATE_KEY) / BUZZ_RELAY_URL / BUZZ_AUTH_TAG — fail-closed
agent = BuzzAgent(kit, [source], exit_after_inactivity=7200)
cause = await agent.run() # blocks until the owner, a signal, or idleness stops it
```
Runnable example: `examples/buzz_agent.py`.
Buzz huddles (live voice calls in ephemeral channels) are handled by the voice subsystem, not this transport: `BuzzHuddleBackend` (`roomkit.voice.backends.buzz_huddle`) is a `VoiceBackend` carrying huddle Opus audio for a `RealtimeVoiceChannel`, and `BuzzHuddleWatcher` owns the announcement-to-call lifecycle, watching the parent channel for kind-48100 announcements (`KIND_HUDDLE_STARTED` / `huddle_announcement_parser` in `roomkit.sources.buzz`) and bridging each huddle. See `examples/buzz_voice_agent.py`.
## Microsoft Teams
```python
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.
```python
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)
```python
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:
```python
from roomkit import WebSocketChannel
ws = WebSocketChannel("ws-client")
# Register a connection — room_id says which conversation this socket is for
ws.register_connection("conn-1", on_receive_callback, room_id="room-1")
# In production, connect to the framework
await kit.connect_websocket("ws-client", "conn-1", send_fn, room_id="room-1")
await kit.disconnect_websocket("ws-client", "conn-1")
```
One channel instance can serve several rooms, so a connection has to say which
one it belongs to: the channel delivers a room's events only to the
connections registered for it. A client holding several conversations open on
one socket subscribes to the extra rooms rather than opening more sockets:
```python
kit.subscribe_websocket("ws-client", "conn-1", "room-2")
kit.unsubscribe_websocket("ws-client", "conn-1", "room-2")
```
## Phone Number Utilities
```python
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:
```python
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
```python
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
```python
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
```python
# 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
```python
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
```python
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_DELTA` | AI is composing a tool call (streaming only) |
| `TOOL_CALL_START` | AI started executing a tool |
| `TOOL_CALL_END` | AI finished executing a tool |
| `CUSTOM` | Custom ephemeral event |
AI channels publish `TOOL_CALL_DELTA` with the tool name and cumulative
`arguments_chars`, never the argument content. A terminal frame with an empty
`tool_calls` closes every composition attempt, including cancellation, provider
failure, retry, and fallback; counts restart for a retry. The complete call
still arrives in `TOOL_CALL_START`.
### Read Tracking
```python
# 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`:
```python
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:
```python
from roomkit import RoomKit, 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:
```python
from roomkit import RoomKit, RateLimit
kit = RoomKit(inbound_rate_limit=RateLimit(max_per_second=10.0))
```
## Circuit Breaker
Isolate provider failures with circuit breakers:
```python
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:
```python
from roomkit 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:
```python
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.
```python
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:
```python
from roomkit import RoomKit, RoomTimers
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:
```python
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
```python
from roomkit import RoomKit, 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
```python
@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. Choose
`InMemoryStore` for ephemeral development state, `SQLiteStore` for an embedded
single-process deployment, or `PostgresStore` when several processes share the
same conversations.
## InMemoryStore (Default)
```python
from roomkit import RoomKit
kit = RoomKit() # Uses InMemoryStore automatically
```
Data lives in Python dicts — fast for development, lost on restart.
## SQLiteStore
`SQLiteStore` is included in the core package and uses only Python's stdlib
`sqlite3`. It is suited to desktop applications, edge deployments, and small
bots running in one process.
```python
from roomkit import RoomKit, SQLiteStore
kit = RoomKit(store=SQLiteStore("roomkit.db"))
```
RoomKit closes the database when `await kit.close()` runs. SQLite history is
stored as Pydantic JSON plus indexed query columns, and message text can be
searched with `await kit.store.search_events("invoice", room_id="support")`.
Do not share one SQLite file between RoomKit worker processes: SQLite
serializes database writes, but it cannot serialize the complete inbound
pipeline and its idempotency decisions. Use PostgreSQL with a distributed lock
manager for horizontal scaling.
## PostgresStore
Install: `pip install roomkit[postgres]`
```python
from roomkit import RoomKit
from roomkit.store.postgres import PostgresStore
store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init() # Creates connection pool and tables
kit = RoomKit(store=store)
```
### Connection Pooling
```python
store = PostgresStore("postgresql://user:pass@localhost/roomkit")
await store.init(min_size=5, max_size=20) # Pool sizing via init()
```
### Schema
PostgresStore creates 11 tables:
| Table | Purpose |
|-------|---------|
| `rooms` | Room records with status, metadata, timers |
| `events` | Event timeline with sequential indexing |
| `event_sequences` | Per-room monotonic event-index high-water marks |
| `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 |
| `identity_addresses` | Multi-channel identity address lookup |
| `read_markers` | Per-channel read positions |
| `schema_version` | Schema migration tracking |
### Operations
```python
# 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
```python
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.):
```python
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
```bash
pip install roomkit[dev]
uv run pytest # Run all tests
uv run pytest tests/test_framework.py -v # Specific file
```
Configuration in `pyproject.toml`:
```toml
[tool.pytest.ini_options]
testpaths = ["tests"]
asyncio_mode = "auto" # No @pytest.mark.asyncio needed
```
## Basic Test Pattern
```python
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
```python
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
```python
from roomkit.voice.backends.mock import MockVoiceBackend
backend = MockVoiceBackend()
# Simulate audio input
backend.simulate_audio_received(session, audio_frame)
```
## Voice Test Bench
`roomkit.voice.testing` replaces frame-by-frame injection and `asyncio.sleep`
with a simulated phone and a hook timeline. `ScenarioVoiceBackend` extends
`MockVoiceBackend`: `play()` cuts a WAV or a `PCMAudio` clip into 20 ms frames
delivered at a transport's cadence (or back to back with `realtime=False`) and
captures what the bot sends per session. `VoiceTrace` records when each voice
hook fired, so a test waits for the hook it needs and reads the turn's order and
latencies off the timeline.
```python
from roomkit import HookTrigger, RoomKit, ScenarioVoiceBackend, VoiceTrace
from roomkit.voice.testing import read_wav, tone, write_wav
backend = ScenarioVoiceBackend()
kit = RoomKit(stt=stt, tts=tts, voice=backend)
trace = VoiceTrace(kit) # before the first session
# ... register the VoiceChannel, create the room, join the session ...
await backend.play(session, tone(300), realtime=False) # or a WAV path
heard = await trace.wait_for(HookTrigger.ON_TRANSCRIPTION, timeout=2)
await trace.wait_for(HookTrigger.AFTER_TTS) # TimeoutError names what did fire
speech_end = trace.first(HookTrigger.ON_SPEECH_END)
assert speech_end is not None
trace.elapsed_ms(speech_end, heard) # STT latency, ms
trace.sequence() # the hooks, in order
backend.write_capture(session, "reports/bot.wav") # what the bot said
```
`read_wav`, `write_wav`, `pcm_frames`, `silence` and `tone` are the stdlib WAV
and PCM helpers around `PCMAudio`. Example: `examples/voice_scenario_backend.py`.
## Mock Pipeline Providers
Every pipeline stage has a mock that accepts pre-configured event sequences:
```python
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
```python
from roomkit.voice.stt.mock import MockSTTProvider
from roomkit.voice.tts.mock import MockTTSProvider
stt = MockSTTProvider(transcripts=["Hello", "How are you?"])
tts = MockTTSProvider()
```
## Mock Identity Resolver
```python
from roomkit.identity.mock import MockIdentityResolver
resolver = MockIdentityResolver()
kit = RoomKit(identity_resolver=resolver)
```
## Mock Realtime Provider
```python
from roomkit.voice.realtime.mock import MockRealtimeProvider, MockRealtimeTransport
provider = MockRealtimeProvider()
transport = MockRealtimeTransport()
```
## Testing Hooks
```python
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
```python
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(transcripts=["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
```python
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
```python
# Pydantic model updates (immutable)
modified = event.model_copy(update={"content": TextContent(body="new")})
# Never mutate models directly — always use model_copy
```
---
RoomKit exports **192 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 |
| `RoomKitConsole` | Full-screen terminal dashboard for voice agent development (optional, requires `rich`) |
| `__version__` | Package version string |
| `content_logging_enabled` | Whether raw message content may be written to logs (default False) |
| `set_content_logging` | Enable/disable process-wide logging of raw message content |
### Channels
| Symbol | Description |
|--------|-------------|
| `ACPChannel` | Connects a room to an external ACP coding agent over stdio |
| `Agent` | AI agent with role, description, greeting, tools |
| `AIChannel` | Intelligence layer for AI responses |
| `AIChannelTurnConfig` | Per-turn generation overrides for AIChannel (None fields keep channel defaults) |
| `AudioVideoChannel` | Combined audio + video channel |
| `BuzzChannel` | Buzz (Nostr relay) transport channel factory |
| `Channel` | Base class for all channels |
| `CLIChannel` | Interactive terminal channel |
| `ConferenceChannel` | Multi-party conference channel backed by an external SFU |
| `DiscordChannel` | Discord bot transport channel factory |
| `EmailChannel` | Email transport channel factory |
| `FrameworkAwareChannel` | Channel base handed the framework it is registered with |
| `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 |
### Conference
| Symbol | Description |
|--------|-------------|
| `BotSession` | The framework's own connection to a conference |
| `ConferenceAccess` | Credentials a client uses to join the conference directly |
| `ConferenceBackend` | ABC for SFU conference backends |
| `ConferenceBargeIn` | Event: a participant spoke over the bot and was allowed to interrupt it |
| `ConferenceCapability` | Flag enum of capabilities a ConferenceBackend can support |
| `ConferenceGrants` | Permissions encoded into a participant's conference access |
| `ConferenceInterruptionConfig` | Multi-party interruption policy |
| `ConferenceInterruptionScope` | Who may interrupt the bot while it is speaking |
| `ConferenceParticipant` | A participant's media presence in a conference |
| `ConferenceRealtimeConfig` | Composes a speech-to-speech provider with a conference |
| `ConferenceRecordingConfig` | Configuration for recording a conference |
| `ConferenceRecordingMode` | Where a conference recording is produced |
| `ConferenceRecordingStarted` | Event: a track's recording has opened |
| `ConferenceRecordingStopped` | Event: a track's recording has closed, with its destination |
| `ConferenceToolHandler` | Callable type for conference tool invocation |
| `ConferenceTrack` | A single media stream published by a conference participant |
| `ConferenceTranscription` | What a lane produced, before it enters the room |
| `TrackKind` | Kind of media carried by a conference track |
| `LiveKitConferenceBackend` | ConferenceBackend backed by a LiveKit SFU |
| `LiveKitConfig` | Connection and behaviour settings for LiveKitConferenceBackend |
| `MockConferenceBackend` | Conference backend that scripts SFU events for tests |
| `MockDelivery` | Mock media timing — how long one frame took to reach every subscriber |
| `MockFaults` | Per-operation failures and delays for the mock backend |
| `MockTrackFormat` | Audio format a participant negotiated for one track (mock) |
| `MockUtterance` | Chunks published for one utterance on one bot's track (mock) |
| `CONFERENCE_ADDRESS_KEYS` | Participant-attribute keys read as a caller's address, most specific first |
| `CONFERENCE_METADATA_KEY` | `Participant.metadata` key a conference nests provider data under (`"conference"`) |
| `CONFERENCE_UNASSERTED_METADATA_KEY` | Metadata key nesting client-claimed (unverified) participant attributes |
### Video
| Symbol | Description |
|--------|-------------|
| `VideoDetectionEvent` | Detection event emitted by video pipeline filters |
| `FaceTouchFilter` | Detects hand-to-face contact using MediaPipe landmarks |
| `FaceTouchConfig` | Configuration for face touch detection |
| `FaceTouchSensitivity` | Sensitivity presets controlling detection thresholds |
| `FaceZone` | Face zones that can be monitored for touch detection |
| `MockFaceTouchFilter` | Mock filter emitting pre-configured detection events at specific frames |
### Enums
| Symbol | Description |
|--------|-------------|
| `Access` | Channel access levels: READ_WRITE, READ_ONLY, WRITE_ONLY, NONE |
| `ChannelCategory` | TRANSPORT or INTELLIGENCE |
| `ChannelType` | 23 values: SMS, MMS, RCS, EMAIL, WHATSAPP, WHATSAPP_PERSONAL, WEBSOCKET, AI, VOICE, REALTIME_VOICE, REALTIME_AUDIO_VIDEO, PUSH, MESSENGER, TELEGRAM, TEAMS, DISCORD, BUZZ, WEBHOOK, VIDEO, AUDIO_VIDEO, CONFERENCE, CLI, SYSTEM |
| `EventStatus` | PENDING, DELIVERED, READ, FAILED, BLOCKED |
| `EventType` | 27 values (MESSAGE, SYSTEM, EDIT, DELETE, TOOL_CALL_START, DTMF, etc.) |
| `HookExecution` | SYNC or ASYNC |
| `HookTrigger` | 76 hook triggers — full list in hooks.md |
| `RoomStatus` | ACTIVE, PAUSED, CLOSED, ARCHIVED |
| `Visibility` | Scope keywords for an event's `visibility` field: ALL, NONE, TRANSPORT, INTELLIGENCE, INTERNAL |
### Models
| Symbol | Description |
|--------|-------------|
| `ChannelBinding` | Binding of a channel to a room |
| `ChannelCapabilities` | Declared capabilities of a channel |
| `ChannelOutput` | Output of a channel delivery |
| `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` | Inbound outcome; `response_metadata` contains the final root intelligence response record |
| `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 |
| `get_current_voice_session` | Get the current voice session from context |
### Tools, Callbacks & Human Input
| Symbol | Description |
|--------|-------------|
| `Tool` | Base class for tool definitions |
| `ToolHandler` | Tool handler type for realtime voice |
| `ToolPolicy` | Per-agent allow/deny rules for tool access |
| `RoleOverride` | Per-role tool policy override |
| `ToolCallCallback` | Callback type for tool call events |
| `ToolCallEvent` | Tool call event model |
| `ToolCallContent` | Content for TOOL_CALL_START and TOOL_CALL_END events |
| `AIGenerationEvent` | Payload for BEFORE_AI_GENERATION hooks, before AI provider invocation |
| `AIResponseEvent` | Payload for ON_AI_RESPONSE hooks, after AI generation completes |
| `BeforeGenerationCallback` | Async callback type receiving AIGenerationEvent |
| `AfterResponseCallback` | Async callback type receiving AIResponseEvent |
| `HumanInputHandler` | Manages pending human input requests |
| `HumanInputToolHandler` | ToolHandler wrapper that blocks on human input for specified tools |
| `PendingInput` | A pending human input request |
| `PendingInputEvent` | Event fired through ON_USER_INPUT_REQUIRED hooks |
| `PendingInputStatus` | Status of a pending human input request |
### Delivery
| Symbol | Description |
|--------|-------------|
| `DeliveryStrategy` | ABC controlling when and how content is delivered to a channel |
| `Immediate` | Deliver now; may interrupt ongoing TTS playback |
| `Queued` | Add to queue, deliver at the next idle window |
| `WaitForIdle` | Wait for TTS/speech to finish, then send |
| `DeliveryBackend` | ABC for persistent delivery queue backends |
| `DeliveryItem` | Serializable delivery request — the unit of work in the queue |
| `DeliveryItemStatus` | Lifecycle status of a delivery item |
| `InMemoryDeliveryBackend` | Asyncio-queue delivery backend (single process, no persistence) |
| `DeliveryResult` | Result of delivering a message |
| `DeliveryStatus` | Delivery status from provider webhook |
### Storage & Locking
| Symbol | Description |
|--------|-------------|
| `ConversationStore` | ABC for persistent room/event/binding/participant storage |
| `InMemoryStore` | Dict-based in-memory store for development and testing |
| `SQLiteStore` | Embedded stdlib SQLite store for single-process persistence |
| `SQLiteSchemaError` | Unsupported or unsafe SQLite schema migration error |
| `RoomLockManager` | ABC for per-room locking |
| `InMemoryLockManager` | In-process per-room asyncio locks with LRU eviction |
| `EventFilter` | Filter criteria for querying room events |
| `PersistencePolicy` | Controls which event types are persisted to the store |
### 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 |
| `ConversationPhase` | Built-in conversation phases (StrEnum) |
| `ConversationState` | Tracks conversation progress within a room |
| `ConversationRouter` | Routes events to the appropriate agent |
| `ConversationPipeline` | Generates routing rules for sequential agent workflows |
| `PipelineStage` | A stage in a ConversationPipeline |
| `RoutingRule` | Routing rule mapping conditions to an agent |
| `RoutingConditions` | Conditions for a routing rule to match |
| `get_conversation_state` | Extract typed ConversationState from room metadata |
| `set_conversation_state` | Return a room copy with updated conversation state |
| `HandoffHandler` | Processes handoff tool calls |
| `HandoffRequest` | Parsed from an agent's handoff tool call arguments |
| `HandoffResult` | Result returned to the calling agent after a handoff |
| `HANDOFF_TOOL` | AITool definition for transferring a conversation to another agent |
| `setup_handoff` | Wires handoff into an AIChannel's tool chain |
### Memory, Skills & Sandbox
| Symbol | Description |
|--------|-------------|
| `MemoryProvider` | ABC for pluggable memory backends feeding AI context construction |
| `Skill` | Full skill definition including instructions body |
| `SkillMetadata` | Lightweight metadata parsed from SKILL.md frontmatter |
| `SkillRegistry` | Discovers, loads, and manages Agent Skills; tracks three visibility states — available, unlisted (`mark_unlisted`: activatable but out of the prompt manifest), unavailable (`mark_unavailable`: listed with a reason, not activatable) |
| `ScriptExecutor` | ABC for executing skill scripts with integrator-defined policy |
| `SandboxExecutor` | ABC for executing commands in a sandboxed environment |
| `SandboxResult` | Result of executing a sandbox command |
### Errors
| Symbol | Description |
|--------|-------------|
| `RoomKitError` | Base exception |
| `RoomNotFoundError` | Room does not exist |
| `RoomClosedError` | Room's status refuses new events (RFC §5.1) |
| `RoomNotAttachedError` | Channel acted on a room it is no longer attached to |
| `ChannelNotFoundError` | Channel not attached to room |
| `ChannelNotRegisteredError` | Channel not registered with framework |
| `ParticipantNotFoundError` | Participant not found in room |
| `ParticipantNotAdmittedError` | Participant barred from what was asked for them |
| `IdentityNotFoundError` | Identity not found |
| `SourceAlreadyAttachedError` | Source already attached |
| `SourceNotFoundError` | No source attached |
| `VoiceBackendNotConfiguredError` | Voice backend not configured |
| `VoiceNotConfiguredError` | Voice (STT/TTS) not configured |
| `ConferenceAlreadyAttachedError` | Second conference channel attached to a room |
| `ConferenceCapabilityError` | Conference operation needs a capability the backend lacks |
| `ConferenceCloseError` | Conference channel did not close all of its resources |
### 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
```python
kit = RoomKit(
store=None, # ConversationStore (default: InMemoryStore)
identity_resolver=None, # IdentityResolver for identifying inbound senders
identity_channel_types=None, # Restrict identity resolution to these ChannelTypes (None = all)
inbound_router=None, # InboundRoomRouter (default: DefaultInboundRoomRouter)
lock_manager=None, # RoomLockManager (default: InMemoryLockManager)
realtime=None, # RealtimeBackend for ephemeral events (default: InMemoryRealtime)
max_chain_depth=5, # Max reentry chain depth — AI-to-AI loop prevention
identity_timeout=10.0, # Identity resolution timeout (seconds)
process_timeout=30.0, # Locked inbound processing timeout (seconds)
stt=None, # STTProvider for transcription
tts=None, # TTSProvider for synthesis
voice=None, # VoiceBackend for real-time audio transport
task_runner=None, # TaskRunner for delegated background tasks (default: InMemoryTaskRunner)
delivery_strategy=None, # DeliveryStrategy | str — proactive delivery of task results
delivery_backend=None, # DeliveryBackend — persistent queue for deliver() (None = in-process)
status_bus=None, # StatusBus for multi-agent coordination (default: in-memory)
telemetry=None, # TelemetryProvider or TelemetryConfig (default: no-op)
inbound_rate_limit=None, # RateLimit applied to inbound messages, keyed per channel_id
orchestration=None, # Default Orchestration strategy for create_room()
persistence_policy=None, # PersistencePolicy — which event types are persisted (None = all)
)
```
## 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?, defer_delivery=False)` | Process an inbound message; deferred results are backfilled by `DeliveryHandle.wait()` |
| `regenerate_response(room_id, *, trigger_id=None)` | Re-run the intelligence channel on the newest accepted transport message; `InboundResult(blocked=True, reason="room_closed")` on a closed or archived room; with `trigger_id`, `reason="trigger_moved"` when the selection under the lock is no longer that event |
| `regenerate_target(room_id)` | The event `regenerate_response` would replay, or `None`; raises `RoomClosedError` when the room refuses events |
### 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
```python
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
```python
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)
```python
from roomkit.voice import (
get_deepgram_provider, get_deepgram_config,
get_elevenlabs_provider, get_elevenlabs_config,
get_gemini_tts_provider, get_gemini_tts_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
```python
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
from roomkit.voice.testing import ScenarioVoiceBackend, VoiceTrace, PCMAudio, read_wav, write_wav
```
### Pipeline
```python
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.stt.language import STTLanguageLock # also exported from roomkit
from roomkit.voice.audio_frame import AudioFrame
```
### Orchestration
```python
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
```python
from roomkit.store.base import ConversationStore
from roomkit.store.memory import InMemoryStore
from roomkit.store.postgres import PostgresStore
```
### Content Types
```python
from roomkit.models.event import (
TextContent, RichContent, MediaContent, AudioContent, VideoContent,
LocationContent, CompositeContent, TemplateContent, SystemContent,
EditContent, DeleteContent,
)
```