# 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, no required dependency beyond Pydantic and PyYAML. > Minimum version: 0.103.0 | 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 (85 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 two dependencies: `pydantic>=2.9`, and `pyyaml>=6`, which reads skill frontmatter. ## 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, room_id="demo-room") ws_bob.register_connection("bob-conn", bob_recv, room_id="demo-room") # 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_stream=…), stored segment by segment 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, room_id="chat") 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") # the channel's own category (TRANSPORT) 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`. ## Cancelling a turn Cancelling a task awaiting `kit.process_inbound()` cancels and drains that call's tools, generation and stream consumers before propagating `asyncio.CancelledError`. Other rooms and shared providers remain usable. Committed events and already streamed text remain in the timeline. With `defer_delivery=True`, explicitly call `await result.delivery.cancel(reason="hangup", timeout=5.0)` to stop that turn. The returned result's `cancellation_reason` names its terminal cancellation; `await result.delivery.wait()` joins the same cleanup. Cancelling a `wait()` alone leaves deferred work running. A completed handle is unchanged by cancel. The default cleanup budget is five seconds on both paths. `TimeoutError` means cleanup is incomplete: keep resources still in use, and for a deferred call await its handle later. Repeated cancellation does not interrupt finalizers. A handle cannot drain its own room lock or lane (`RuntimeError`). External effects already committed and tasks detached by application handlers are outside this cancellation contract. See `examples/cancel_delivery.py`. --- 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. **Three triggers invert that**, because their payload is content a hook may exist to withhold, or an action it may exist to prevent: | Trigger | On hook failure | |---|---| | `BEFORE_TTS` | **Blocked** — the text is not synthesised | | `ON_TRANSCRIPTION` | **Blocked** — the transcript is not published | | `BEFORE_TOOL_USE` | **Refused** — the call does not run; the model reads `Tool 'x' denied by pre-execution hook.`, and the hook's error reaches ON_TOOL_CALL's observers on `error_detail` (an external tool handler decides and reports the call itself) | | every other sync trigger | Allowed, error logged | All four failure modes block on those three, 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 | | `ON_VERIFICATION_REQUIRED` | SYNC | `(event, ctx) -> HookResult` | An identified sender who is not verified wrote, under a `verification_policy` (RFC §11.7). `allow()` holds the message from the agents, `allow_unverified()` passes it, `block()` refuses it. | | `ON_VERIFICATION_COMPLETED` | ASYNC | `(event, ctx) -> None` | A verification request was completed. A SYSTEM event whose data carries `identity_id`, `organization_id`, `request_id`, `channel_id`, `method`, `expires_at`. | | `ON_VERIFICATION_ENDED` | ASYNC | `(event, ctx) -> None` | A verification or an open request ended: expired, revoked, cancelled or failed. Same data plus `reason` and `ended` (`"verification"` or `"request"`). | ### 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` | SYNC | `(event, ctx) -> HookResult` | Before a proactive `kit.deliver()` strategy executes (in-process and worker paths), a delegation's or a supervisor's background result included. Payload is a synthetic INTERNAL system-source event describing the delivery, of type `INSTRUCTION` for `deliver(instruction=True)`. `HookResult.block()` refuses the delivery (outcome `blocked`); a modified event's text replaces the content. A hook that raises does not block (fail-open, RFC §9.3). | | `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**. On a streamed response it runs once per sentence: a block drops that sentence. | | `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` | Recording started (voice session, conference track or room), before any media: the consent point | | `ON_RECORDING_STOPPED` | ASYNC | `(RecordingStoppedEvent, ctx) -> None` | Recording stopped (voice session, conference track or room), result available; a room's `session` is `None` and `room_id` names it | | `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, conference, external/ACP tools). Block denies the call, and the model reads the block's reason. **Fails closed**: a hook that raises, times out or returns something unusable denies the call; the model then reads `Tool 'x' denied by pre-execution hook.`, never the hook's error, which reaches ON_TOOL_CALL's observers on `error_detail` (an external tool handler decides and reports the call itself). A room context that cannot be built refuses the call too. | | `ON_TOOL_CALL` | SYNC | `(ToolCallEvent, ctx) -> HookResult` | A tool call executed (AIChannel, RealtimeVoiceChannel, skills, external/ACP tools). In the AI channel's tool loop, Block returns an error result to the model and marks the call failed, its structured copy dropped; `HookResult(metadata={"result": ...})` or `HookResult.modify(event)` supplies or overrides the tool result (`event.result is None` means the hook must provide it; a served result replaced with `None` reads as JSON `null`, on every channel) and keeps the copy; SYNC hooks chain, each seeing the result the previous one left, and ASYNC observers see the final one as the model reads it (a replacement serialized), a blocked call included (`is_error=True`), while a report on a call whose outcome the model already read (an external tool's, one the provider ran itself) reaches them as it was read, whatever a SYNC hook returned; `event.structured_content` is the copy the tool-call event carries for UI surfaces (MCP `structuredContent`), which `metadata={"structured_content": ...}` replaces and `None` clears. RFC §9 lists this ASYNC; code invokes it through the sync pipeline. | | `AFTER_TOOL_ROUND` | SYNC | `(ToolRoundEvent, ctx) -> HookResult` | Between two rounds of an AIChannel's tool loop, after each round the channel ran calls in. The event carries the round whole (`calls`, the channel's `results`, the provider's `answered`) and the names of the turn's toolset (`tools`). Act on it in place: `event.withdraw(*names)` takes tools out of the rest of the turn with every guarantee of a BEFORE_AI_GENERATION withdrawal, `event.add_message(text)` is what the next round reads after the results. A BLOCK stops the hooks after it and changes nothing of the round; a MODIFY's event is not read. Not fired by realtime sessions, a reasoning backend's agent or an ACP agent's own loop. | | `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 | | `ON_REALTIME_DELEGATION` | ASYNC | `(RealtimeDelegationEvent, ctx) -> None` | A full-duplex model handed reasoning or tool use to a backend, hosted or integrator-side | ### 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. Its `tools` are the turn's toolset after the tool policy and skill gating, under Tool Search the whole catalogue: a tool it removes is declared at no round, named by no `find_tools`, and refused; a tool it adds is declared at every round, never deferred. | | `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`, `budget_exceeded`, `cancelled`, `force_stopped`, `truncated`, `empty_response`, `unfinished`, `error`). 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`. | ### Speaking Turns | Trigger | Execution | Signature | Description | |---------|-----------|-----------|-------------| | `ON_SPEAK_DECISION` | ASYNC | `(SpeakDecisionEvent, ctx) -> None` | An AI channel's speak policy decided whether the agent speaks, offers or stays silent on an event: the decision with its reason and judgments, `duration_ms`, `asked_again` when it decided again once the agent thought. | | `ON_THOUGHT` | ASYNC | `(ThoughtEvent, ctx) -> None` | An AI channel's thinker came back with a new thought in a room, or speaking emptied what the agent wanted to say: the thought, the one it replaces, `duration_ms` of the thinker call. | ### Discussion | Trigger | Execution | Signature | Description | |---------|-----------|-----------|-------------| | `ON_DISPATCH_DECISION` | ASYNC | `(DispatchDecisionEvent, ctx) -> None` | A discussion's dispatch policy decided who takes a person's message that names no agent (RFC §19.7.5 rule 18): `event.candidates`, `event.decision` (`agents` in order, `reason`, `judgments`), `event.duration_ms`. The reason `fallback` means the policy failed or ran past `dispatch_timeout`, and every candidate was asked. | | `ON_SPEAK_QUEUE` | ASYNC | `(SpeakQueueEvent, ctx) -> None` | A discussion's speak queue changed (RFC §19.7.5): `event.change` is a `SpeakQueueChange` (`queued`, `turn_given`, `turn_ended`, `instruction_dropped`, `listening`, `talking_again`, `waiting`, `over`), `event.channel_ids` the agents it concerns, `event.queue` the `SpeakQueue` as it now is. Announced in the order the changes happened. | ### 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; the one addition is billing labels: `GeminiVertexConfig.labels` (`{"tenant": "acme"}`) rides every request as `GenerateContentConfig.labels`, so Cloud Billing splits one project's spend per tenant or partner: metadata only, 24 to 48 h behind in the report, never a source of truth; Google's label rules (64 labels, 63 characters, lowercase letters, digits, `_`, `-`, a key starts with a letter) are enforced when the config is built, and the field is Vertex-only because the Developer API refuses it. - **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. - **Outbound policy** (`transport=`) — `OpenAIAIProvider(config, transport=...)`, and every provider built on it (`AzureAIProvider`, `OpenRouterAIProvider`, `create_vllm_provider(config, transport=...)`, the inherited xAI, DeepSeek, Qwen, LiteLLM, Cerebras, Meta), sends every request through an `httpx.AsyncBaseTransport` placed inside the SDK's own default client (redirects followed, each hop through the transport; per-request timeout kept). The transport owns its connection pool and limits, and httpx reads no environment proxy beside a transport. The seam for a policy that judges the address actually dialled when users name the endpoint, or a `MockTransport` in a test; closed with the provider. See `examples/openai_outbound_policy.py`. - **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-3.8-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?", ) ``` `role`, `description`, `scope` and `language` are appended to the system prompt as an `--- Agent Identity ---` block, in a turn, a handoff and a realtime pipeline alike. A host that renders the agent's identity in its own prompt passes `identity_in_prompt=False`: no block is written anywhere, and the fields stay readable on the agent. ## Steering a Running Turn `steer()` hands a running tool loop a directive (`Cancel`, `InjectMessage`, `UpdateSystemPrompt`, RFC §21.3) and returns how many loops it reached. One channel object serves every room it is bound to, so a host acting for one room addresses that room: ```python from roomkit.models.steering import Cancel, InjectMessage agent.steer(Cancel(reason="stop pressed"), room_id="room-a") # every loop of room-a agent.steer(InjectMessage(content="Also check the logs."), room_id="room-a") # its latest agent.steer(Cancel(reason="timeout"), loop_id="loop-abc123") # one loop ``` Without `loop_id` or `room_id`, a directive reaches the channel's most recent loop, whatever its room. A loop is reachable once its turn has started (its response stream is read): a directive that comes before reaches none, and `steer()` returns 0. See `examples/ai_shared_agent_rooms.py`. ## 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, ) ``` #### Refusing a call What the handler returns is the tool's answer, so a refusal returned as a body reads as work that was done: the tool-call event says `completed`, and an audit trail records a call that never ran. Raise `ToolRefusedError` instead. ```python from roomkit import ToolRefusedError async def handle_tools(name: str, arguments: dict) -> str: if name not in MY_TOOLS: raise ToolRefusedError(f"Error: the tool '{name}' does not exist.") ... ``` The message reaches the model verbatim, which is the point: any other exception reads as `{"error": "Tool '' failed ()"}`, its message withheld from the model (it can hold a password or a path) and handed to the log and to `ON_TOOL_CALL` observers as `event.error_detail`. The call is marked failed, observers see `event.is_error`, and the stored `TOOL_CALL_END` carries what the model read. A tool that ran and failed, with words for the model, raises `ToolFailedError`: the message reaches the model verbatim and the observers as `error_detail`, and the call is failed, not refused (`event.refused` is false). An MCP tool whose result says `isError` raises it. ### 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 async with MCPToolProvider.from_command("uvx", ["mcp-server-sqlite", "--db", "data.db"]) as mcp: ai = AIChannel("ai", provider=provider, tools=mcp.get_tools(), tool_handler=mcp.as_tool_handler()) ``` Beside the model's tools, a host reads what an MCP App needs from the same connection: `mcp.tool_meta()` (each discovered tool's `_meta`, `ui.resourceUri` and `ui.csp` among it, from the listing made at connection), `await mcp.read_resource(uri)` (the server's `ReadResourceResult`, the app's HTML) and `await mcp.call_tool_result(name, arguments)` (the server's `CallToolResult` as it is, `isError` included, for a frame's own calls). Like `call_tool`, `call_tool_result` calls any tool the server has: `tool_filter` shapes what discovery offers the model, not what the host may call, so the host authorizes a frame's call itself. #### Server instructions A server may publish, when it connects, how to use its tools (`InitializeResult.instructions`). `mcp.instructions` keeps that text (`None` when the server sent none) and `mcp.server_info` what the server says of itself (`name`, `version`). Hand the instructions to the channel, under a label of your own for each server: ```python async with MCPToolProvider.from_command("gamemaster-mcp") as gm: ai = AIChannel( "ai", provider=provider, system_prompt="You are the game master.", tools=gm.get_tools(), tool_handler=gm.as_tool_handler(), server_instructions={"gm": gm.instructions}, ) ``` The channel adds them to every system prompt it composes, after its own prompt and before the skills, so a prompt a binding or a `config_provider` sets for the room keeps them, Tool Search does not hide them, and no eviction or compaction reaches them. Each server's text is that server's data: it sits in a `` block it cannot close, under `## MCP server: