Multi-Channel Conversations,
Simplified.

from roomkit import RoomKit, HookTrigger, IdentityHookResult

kit = RoomKit(identity_resolver=my_resolver)

# Handle ambiguous identity (multiple matches)
@kit.identity_hook(HookTrigger.ON_IDENTITY_AMBIGUOUS)
async def resolve_ambiguous(event, ctx, id_result):
    # Access sender info directly
    sender = id_result.address  # e.g., "+14185551234"

    if sender in known_senders:
        identity = get_identity(known_senders[sender])
        return IdentityHookResult.resolved(identity)

    # Ask user to identify themselves
    return IdentityHookResult.pending(
        display_name=f"Unknown ({sender})",
        candidates=id_result.candidates
    )

from roomkit import RoomKit, EphemeralEventType

kit = RoomKit()

# Subscribe to typing indicators and presence
async def on_realtime(event):
    if event.type == EphemeralEventType.TYPING_START:
        print(f"{event.user_id} is typing...")
    elif event.type == EphemeralEventType.PRESENCE_ONLINE:
        print(f"{event.user_id} came online")

sub_id = await kit.subscribe_room("room-123", on_realtime)

# Publish typing indicator
await kit.publish_typing("room-123", "user-456")

# Publish read receipt
await kit.publish_read_receipt("room-123", "user-456", "event-789")

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
from roomkit.voice.realtime.ws_transport import WebSocketRealtimeTransport

# Speech-to-speech AI — no STT/TTS pipeline needed
provider = GeminiLiveProvider(api_key="...")
transport = WebSocketRealtimeTransport()

kit = RoomKit()
kit.register_channel(RealtimeVoiceChannel(
    "realtime-voice",
    provider=provider,
    transport=transport,
    system_prompt="You are a helpful voice assistant.",
))

# Connect a participant — audio flows directly to/from Gemini
session = await channel.start_session("room-1", "user-1", websocket)

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

from roomkit import RoomKit
from roomkit.channels.ai import AIChannel
from roomkit.orchestration import (
    Agent, ConversationPipeline, PipelineStage
)

kit = RoomKit()

# Define agents with distinct roles
triage = Agent(
    "triage", provider=my_ai,
    role="Classify the customer's intent",
    greeting="Hi! How can I help you today?",
)
handler = Agent(
    "handler", provider=my_ai,
    role="Resolve the customer's issue",
)

# Wire into a pipeline
pipeline = ConversationPipeline(stages=[
    PipelineStage(phase="triage", agent_id="triage", next="handling"),
    PipelineStage(phase="handling", agent_id="handler", next=None),
])

# Install — router and handler are wired automatically
router, on_handoff = pipeline.install(kit, [triage, handler])

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
)