Multi-Channel Conversations,
Simplified.

from roomkit import RoomKit, AIChannel, Tool

class GetWeatherTool(Tool):
    @property
    def definition(self) -> dict:
        return {
            "name": "get_weather",
            "description": "Get current weather",
            "parameters": {"type": "object", "properties": {
                "city": {"type": "string"}
            }},
        }

    async def handler(self, name, arguments):
        return fetch_weather(arguments["city"])

# Pass Tool objects directly — definition + handler in one class
ai = AIChannel("ai", provider=my_ai, tools=[GetWeatherTool()])

from roomkit import RoomKit, VoiceChannel
from roomkit.voice.stt.deepgram import DeepgramSTTProvider
from roomkit.voice.tts.elevenlabs import ElevenLabsTTSProvider
from roomkit.voice.backends.fastrtc import FastRTCBackend

# Configure voice channel with STT, TTS, and backend
kit = RoomKit()
kit.register_channel(VoiceChannel(
    "voice",
    stt=DeepgramSTTProvider(api_key="..."),
    tts=ElevenLabsTTSProvider(api_key="...", voice_id="..."),
    backend=FastRTCBackend(),
))

# Attach to a room — voice joins the same conversation
await kit.attach_channel(room.id, "voice")

# Transcriptions flow through hooks like any message

from roomkit import RoomKit, RealtimeVoiceChannel
from roomkit.providers.gemini.realtime import GeminiLiveProvider

# Speech-to-speech AI: no STT/TTS pipeline needed
kit = RoomKit()
kit.register_channel(RealtimeVoiceChannel(
    "realtime",
    provider=GeminiLiveProvider(api_key="..."),
    system_prompt="You are a helpful voice assistant.",
    tools=[GetWeatherTool()],
))

# Join a room: audio flows directly to/from AI
await kit.join(room.id, "realtime", websocket)

# Transcriptions appear as RoomEvents in the room
# Text from other channels is injected into the session

from roomkit import RoomKit, VideoChannel
from roomkit.video.vision.openai import OpenAIVisionProvider
from roomkit.video.backends.fastrtc import FastRTCVideoBackend

kit = RoomKit()
kit.register_channel(VideoChannel(
    "video",
    backend=FastRTCVideoBackend(),
    vision=OpenAIVisionProvider(api_key="..."),
))

# Vision AI analyzes frames on demand
# Filters: YOLO detection, watermark, censor
# Recording: H.264/VP9 with room-level A/V sync

await kit.attach_channel(room.id, "video")

from roomkit import RoomKit, Agent
from roomkit.memory import RetrievalMemory
from roomkit.knowledge.postgres import PostgresKnowledgeSource

# Knowledge-enriched context from pluggable sources
memory = RetrievalMemory(
    sources=[PostgresKnowledgeSource(pool=db_pool)],
    max_results=5,
)

agent = Agent(
    "support", provider=ai,
    role="Customer support agent",
    memory=memory,
)

# Agent context is automatically enriched with
# relevant knowledge from your database

from roomkit import RoomKit, BaseSourceProvider, SourceStatus

class NATSSource(BaseSourceProvider):
    def __init__(self, subject: str):
        super().__init__()
        self.subject = subject

    @property
    def name(self) -> str:
        return f"nats:{self.subject}"

    async def start(self, emit):
        self._set_status(SourceStatus.CONNECTED)
        async for msg in self.subscribe():
            await emit(parse_message(msg))
            self._record_message()

# Attach with resilience options
await kit.attach_source(
    "nats-events", NATSSource("chat.>"),
    max_restart_attempts=10,   # Give up after 10 failures
    max_concurrent_emits=20,   # Backpressure control
)