diff --git a/backend/.env.example b/backend/.env.example index ded015af..c6ae0550 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -17,3 +17,25 @@ APPLE_TEAM_ID=ABCDE12345 # GitHub Releases (required for --publish) # ============================================================================= GH_TOKEN=ghp_your-github-personal-access-token + +# ============================================================================= +# Channels: SMS, WhatsApp, Voice Calling +# ============================================================================= +TWILIO_ACCOUNT_SID=your-twilio-account-sid +TWILIO_AUTH_TOKEN=your-twilio-auth-token +TWILIO_PHONE_NUMBER=+1234567890 + +TELNYX_API_KEY=your-telnyx-api-key +TELNYX_PUBLIC_KEY=your-telnyx-public-key + +# ============================================================================= +# TTS / STT Providers +# ============================================================================= +ELEVENLABS_API_KEY=your-elevenlabs-api-key +DEEPGRAM_API_KEY=your-deepgram-api-key +OPENAI_API_KEY=your-openai-api-key + +# ============================================================================= +# Webhook URL (required for inbound SMS/calls — use ngrok or Tailscale) +# ============================================================================= +WEBHOOK_BASE_URL=https://your-public-url.ngrok.io diff --git a/backend/apps/channels/__init__.py b/backend/apps/channels/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/channels/adapters/__init__.py b/backend/apps/channels/adapters/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/apps/channels/adapters/telnyx_adapter.py b/backend/apps/channels/adapters/telnyx_adapter.py new file mode 100644 index 00000000..e5cf037f --- /dev/null +++ b/backend/apps/channels/adapters/telnyx_adapter.py @@ -0,0 +1,146 @@ +"""Telnyx implementation of BaseChannelAdapter. + +Uses Telnyx Call Control v2 for voice and Messaging API for SMS. +""" +import asyncio +import hashlib +import hmac +import logging +from typing import Optional +from functools import partial + +from backend.apps.channels.base_adapter import BaseChannelAdapter + +logger = logging.getLogger(__name__) + + +class TelnyxAdapter(BaseChannelAdapter): + + def __init__(self, api_key: str, public_key: str = ""): + self._api_key = api_key + self._public_key = public_key + self._telnyx = None + + def _get_telnyx(self): + if self._telnyx is None: + import telnyx + telnyx.api_key = self._api_key + self._telnyx = telnyx + return self._telnyx + + async def send_sms(self, to: str, from_: str, body: str) -> dict: + telnyx = self._get_telnyx() + loop = asyncio.get_event_loop() + msg = await loop.run_in_executor( + None, + partial( + telnyx.Message.create, + to=to, + from_=from_, + text=body, + ), + ) + return {"id": msg.id, "status": getattr(msg, "status", "queued")} + + async def send_whatsapp(self, to: str, from_: str, body: str) -> dict: + # Telnyx WhatsApp uses the same messaging API with messaging_profile_id + return await self.send_sms(to, from_, body) + + async def initiate_call( + self, to: str, from_: str, webhook_url: str, greeting: str = "" + ) -> dict: + telnyx = self._get_telnyx() + loop = asyncio.get_event_loop() + call = await loop.run_in_executor( + None, + partial( + telnyx.Call.create, + to=to, + from_=from_, + connection_id=self._api_key, # connection_id should be set separately + webhook_url=webhook_url, + ), + ) + return {"call_control_id": call.call_control_id, "status": "initiated"} + + def verify_webhook_signature( + self, request_url: str, params: dict, signature: str, auth_token: str + ) -> bool: + if not self._public_key: + # Fail closed: no public key means reject + logger.error("Telnyx public key not configured — rejecting webhook") + return False + try: + # Telnyx uses Ed25519 signature verification + # The signature and timestamp are in webhook headers + import base64 + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + from cryptography.hazmat.primitives.serialization import load_pem_public_key + + public_key = load_pem_public_key(self._public_key.encode()) + sig_bytes = base64.b64decode(signature) + payload = params.get("_raw_body", "") + timestamp = params.get("_timestamp", "") + signed_payload = f"{timestamp}|{payload}" + public_key.verify(sig_bytes, signed_payload.encode()) + return True + except Exception: + logger.exception("Telnyx signature verification failed") + return False + + def generate_twiml_gather( + self, + prompt: str, + action_url: str, + voice: str = "Polly.Joanna", + language: str = "en-US", + timeout: int = 10, + ) -> str: + # Telnyx uses TeXML (Twilio-compatible XML) + return ( + '' + "" + f'' + f'{_escape_xml(prompt)}' + "" + f'I didn\'t hear anything. Goodbye.' + "" + ) + + def generate_twiml_say( + self, text: str, voice: str = "Polly.Joanna", language: str = "en-US" + ) -> str: + return ( + '' + "" + f'{_escape_xml(text)}' + "" + ) + + def generate_twiml_hangup(self) -> str: + return ( + '' + "" + ) + + async def get_recording_audio(self, recording_url: str, auth_token: str) -> bytes: + import httpx + async with httpx.AsyncClient() as client: + resp = await client.get( + recording_url, + headers={"Authorization": f"Bearer {self._api_key}"}, + follow_redirects=True, + ) + resp.raise_for_status() + return resp.content + + +def _escape_xml(text: str) -> str: + return ( + text.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + .replace("'", "'") + ) diff --git a/backend/apps/channels/adapters/twilio_adapter.py b/backend/apps/channels/adapters/twilio_adapter.py new file mode 100644 index 00000000..7246bea1 --- /dev/null +++ b/backend/apps/channels/adapters/twilio_adapter.py @@ -0,0 +1,139 @@ +"""Twilio implementation of BaseChannelAdapter. + +Handles SMS, WhatsApp, and Voice via the Twilio Python SDK. +""" +import asyncio +import logging +from typing import Optional +from functools import partial + +from backend.apps.channels.base_adapter import BaseChannelAdapter + +logger = logging.getLogger(__name__) + + +class TwilioAdapter(BaseChannelAdapter): + + def __init__(self, account_sid: str, auth_token: str): + self._account_sid = account_sid + self._auth_token = auth_token + self._client = None + + def _get_client(self): + if self._client is None: + from twilio.rest import Client + self._client = Client(self._account_sid, self._auth_token) + return self._client + + async def send_sms(self, to: str, from_: str, body: str) -> dict: + client = self._get_client() + loop = asyncio.get_event_loop() + msg = await loop.run_in_executor( + None, + partial( + client.messages.create, + to=to, + from_=from_, + body=body, + ), + ) + return {"sid": msg.sid, "status": msg.status} + + async def send_whatsapp(self, to: str, from_: str, body: str) -> dict: + wa_to = to if to.startswith("whatsapp:") else f"whatsapp:{to}" + wa_from = from_ if from_.startswith("whatsapp:") else f"whatsapp:{from_}" + client = self._get_client() + loop = asyncio.get_event_loop() + msg = await loop.run_in_executor( + None, + partial( + client.messages.create, + to=wa_to, + from_=wa_from, + body=body, + ), + ) + return {"sid": msg.sid, "status": msg.status} + + async def initiate_call( + self, to: str, from_: str, webhook_url: str, greeting: str = "" + ) -> dict: + client = self._get_client() + loop = asyncio.get_event_loop() + call = await loop.run_in_executor( + None, + partial( + client.calls.create, + to=to, + from_=from_, + url=webhook_url, + ), + ) + return {"sid": call.sid, "status": call.status} + + def verify_webhook_signature( + self, request_url: str, params: dict, signature: str, auth_token: str + ) -> bool: + try: + from twilio.request_validator import RequestValidator + validator = RequestValidator(auth_token) + return validator.validate(request_url, params, signature) + except Exception: + logger.exception("Twilio signature verification failed") + return False + + def generate_twiml_gather( + self, + prompt: str, + action_url: str, + voice: str = "Polly.Joanna", + language: str = "en-US", + timeout: int = 10, + ) -> str: + return ( + '' + "" + f'' + f'{_escape_xml(prompt)}' + "" + f'I didn\'t hear anything. Goodbye.' + "" + ) + + def generate_twiml_say( + self, text: str, voice: str = "Polly.Joanna", language: str = "en-US" + ) -> str: + return ( + '' + "" + f'{_escape_xml(text)}' + "" + ) + + def generate_twiml_hangup(self) -> str: + return ( + '' + "" + ) + + async def get_recording_audio(self, recording_url: str, auth_token: str) -> bytes: + import httpx + async with httpx.AsyncClient() as client: + resp = await client.get( + recording_url, + auth=(self._account_sid, auth_token), + follow_redirects=True, + ) + resp.raise_for_status() + return resp.content + + +def _escape_xml(text: str) -> str: + return ( + text.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + .replace("'", "'") + ) diff --git a/backend/apps/channels/base_adapter.py b/backend/apps/channels/base_adapter.py new file mode 100644 index 00000000..bafeeac7 --- /dev/null +++ b/backend/apps/channels/base_adapter.py @@ -0,0 +1,88 @@ +from abc import ABC, abstractmethod +from typing import Optional + + +class BaseChannelAdapter(ABC): + """Provider-agnostic interface for telephony operations.""" + + @abstractmethod + async def send_sms(self, to: str, from_: str, body: str) -> dict: + """Send an SMS message. Returns provider response dict.""" + ... + + @abstractmethod + async def send_whatsapp(self, to: str, from_: str, body: str) -> dict: + """Send a WhatsApp message. Returns provider response dict.""" + ... + + @abstractmethod + async def initiate_call( + self, to: str, from_: str, webhook_url: str, greeting: str = "" + ) -> dict: + """Initiate an outbound voice call. Returns provider response dict.""" + ... + + @abstractmethod + def verify_webhook_signature( + self, request_url: str, params: dict, signature: str, auth_token: str + ) -> bool: + """Verify that an inbound webhook is authentic.""" + ... + + @abstractmethod + def generate_twiml_gather( + self, + prompt: str, + action_url: str, + voice: str = "Polly.Joanna", + language: str = "en-US", + timeout: int = 10, + ) -> str: + """Generate TwiML (or equivalent) to play a prompt and gather speech.""" + ... + + @abstractmethod + def generate_twiml_say( + self, text: str, voice: str = "Polly.Joanna", language: str = "en-US" + ) -> str: + """Generate TwiML (or equivalent) to speak text.""" + ... + + @abstractmethod + def generate_twiml_hangup(self) -> str: + """Generate TwiML (or equivalent) to end a call.""" + ... + + @abstractmethod + async def get_recording_audio(self, recording_url: str, auth_token: str) -> bytes: + """Download audio from a recording URL.""" + ... + + def chunk_message(self, text: str, max_length: int = 1600) -> list[str]: + """Split a long message into chunks respecting sentence boundaries.""" + if len(text) <= max_length: + return [text] + + chunks: list[str] = [] + remaining = text + + while remaining: + if len(remaining) <= max_length: + chunks.append(remaining) + break + + # Try to split at sentence boundary + split_at = -1 + for sep in [". ", "! ", "? ", "\n\n", "\n", " "]: + idx = remaining.rfind(sep, 0, max_length) + if idx > 0: + split_at = idx + len(sep) + break + + if split_at <= 0: + split_at = max_length + + chunks.append(remaining[:split_at].rstrip()) + remaining = remaining[split_at:].lstrip() + + return chunks diff --git a/backend/apps/channels/call_state.py b/backend/apps/channels/call_state.py new file mode 100644 index 00000000..6a3abc95 --- /dev/null +++ b/backend/apps/channels/call_state.py @@ -0,0 +1,129 @@ +import logging +from datetime import datetime +from typing import Optional, Literal + +logger = logging.getLogger(__name__) + +CallStatus = Literal[ + "ringing", "connected", "gathering", "processing", "responding", "completed", "failed" +] + +VALID_TRANSITIONS: dict[CallStatus, set[CallStatus]] = { + "ringing": {"connected", "completed", "failed"}, + "connected": {"gathering", "completed", "failed"}, + "gathering": {"processing", "completed", "failed"}, + "processing": {"responding", "completed", "failed"}, + "responding": {"gathering", "completed", "failed"}, + "completed": set(), + "failed": set(), +} + + +class CallState: + """Tracks the lifecycle of a single voice call.""" + + def __init__( + self, + call_sid: str, + channel_id: str, + from_number: str, + to_number: str, + ): + self.call_sid = call_sid + self.channel_id = channel_id + self.from_number = from_number + self.to_number = to_number + self.agent_session_id: Optional[str] = None + self.status: CallStatus = "ringing" + self.turns: list[dict] = [] + self.created_at = datetime.now() + self.last_activity = datetime.now() + self.error: Optional[str] = None + + def transition(self, new_status: CallStatus) -> bool: + """Attempt a state transition. Returns True if valid.""" + if new_status in VALID_TRANSITIONS.get(self.status, set()): + logger.info( + "Call %s: %s -> %s", self.call_sid, self.status, new_status + ) + self.status = new_status + self.last_activity = datetime.now() + return True + logger.warning( + "Call %s: invalid transition %s -> %s", + self.call_sid, self.status, new_status, + ) + return False + + def add_turn(self, role: str, content: str): + self.turns.append({ + "role": role, + "content": content, + "timestamp": datetime.now().isoformat(), + }) + self.last_activity = datetime.now() + + @property + def is_active(self) -> bool: + return self.status not in ("completed", "failed") + + @property + def duration_seconds(self) -> float: + return (datetime.now() - self.created_at).total_seconds() + + def to_dict(self) -> dict: + return { + "call_sid": self.call_sid, + "channel_id": self.channel_id, + "from_number": self.from_number, + "to_number": self.to_number, + "agent_session_id": self.agent_session_id, + "status": self.status, + "turns": self.turns, + "created_at": self.created_at.isoformat(), + "last_activity": self.last_activity.isoformat(), + "duration_seconds": self.duration_seconds, + "error": self.error, + } + + +class CallManager: + """Manages all active voice calls.""" + + def __init__(self): + self.calls: dict[str, CallState] = {} + + def create_call( + self, + call_sid: str, + channel_id: str, + from_number: str, + to_number: str, + ) -> CallState: + call = CallState(call_sid, channel_id, from_number, to_number) + self.calls[call_sid] = call + return call + + def get_call(self, call_sid: str) -> Optional[CallState]: + return self.calls.get(call_sid) + + def end_call(self, call_sid: str, status: CallStatus = "completed"): + call = self.calls.get(call_sid) + if call: + call.transition(status) + + def cleanup_stale(self, max_duration_seconds: int = 3600): + """Remove calls that have exceeded max duration.""" + stale = [ + sid + for sid, call in self.calls.items() + if not call.is_active or call.duration_seconds > max_duration_seconds + ] + for sid in stale: + if self.calls[sid].is_active: + self.calls[sid].transition("failed") + self.calls[sid].error = "Exceeded max call duration" + del self.calls[sid] + + def get_active_calls(self) -> list[dict]: + return [c.to_dict() for c in self.calls.values() if c.is_active] diff --git a/backend/apps/channels/channels.py b/backend/apps/channels/channels.py new file mode 100644 index 00000000..60da8283 --- /dev/null +++ b/backend/apps/channels/channels.py @@ -0,0 +1,387 @@ +"""Channels SubApp — REST endpoints and Twilio/Telnyx webhooks.""" +import logging +import os +from contextlib import asynccontextmanager +from datetime import datetime + +from fastapi import HTTPException, Request +from fastapi.responses import JSONResponse, Response + +from backend.config.Apps import SubApp +from backend.apps.channels.models import ( + ChannelConfig, ChannelCreate, ChannelUpdate, VoiceConfig, TTSConfig, STTConfig, +) +from backend.apps.channels.orchestrator import channel_orchestrator +from backend.apps.channels import ws_events + +logger = logging.getLogger(__name__) + + +@asynccontextmanager +async def channels_lifespan(): + logger.info("Channels sub-app starting") + await channel_orchestrator.restore_all() + yield + logger.info("Channels sub-app shutting down") + await channel_orchestrator.persist_all() + + +channels = SubApp("channels", channels_lifespan) + + +# ─── CRUD Endpoints ────────────────────────────────────────────── + + +@channels.router.get("/list") +async def list_channels(): + configs = list(channel_orchestrator.configs.values()) + return { + "channels": [c.model_dump(mode="json") for c in configs], + } + + +@channels.router.get("/{channel_id}") +async def get_channel(channel_id: str): + config = channel_orchestrator.configs.get(channel_id) + if not config: + raise HTTPException(404, "Channel not found") + return config.model_dump(mode="json") + + +@channels.router.post("/create") +async def create_channel(body: ChannelCreate): + config = ChannelConfig( + name=body.name, + channel_type=body.channel_type, + provider=body.provider, + phone_number=body.phone_number, + credentials=body.credentials, + ) + if body.agent_config: + config.agent_config = body.agent_config + if body.security: + config.security = body.security + if body.voice_config: + config.voice_config = body.voice_config + if body.tts_config: + config.tts_config = body.tts_config + if body.stt_config: + config.stt_config = body.stt_config + + channel_orchestrator.save_config(config) + return {"channel": config.model_dump(mode="json")} + + +@channels.router.put("/{channel_id}") +async def update_channel(channel_id: str, body: ChannelUpdate): + config = channel_orchestrator.configs.get(channel_id) + if not config: + raise HTTPException(404, "Channel not found") + + updates = body.model_dump(exclude_none=True) + for key, val in updates.items(): + setattr(config, key, val) + + # Re-create adapter if credentials changed + if "credentials" in updates or "provider" in updates: + channel_orchestrator.adapters.pop(channel_id, None) + + channel_orchestrator.save_config(config) + return {"channel": config.model_dump(mode="json")} + + +@channels.router.delete("/{channel_id}") +async def delete_channel(channel_id: str): + if channel_id not in channel_orchestrator.configs: + raise HTTPException(404, "Channel not found") + channel_orchestrator.delete_config(channel_id) + return {"ok": True} + + +# ─── Enable / Disable / Test ───────────────────────────────────── + + +@channels.router.post("/{channel_id}/enable") +async def enable_channel(channel_id: str): + config = channel_orchestrator.configs.get(channel_id) + if not config: + raise HTTPException(404, "Channel not found") + + try: + channel_orchestrator.get_adapter(config) + config.enabled = True + config.status = "active" + config.status_message = None + channel_orchestrator.save_config(config) + await ws_events.emit_channel_status(channel_id, "active") + return {"ok": True, "status": "active"} + except Exception as e: + config.status = "error" + config.status_message = str(e) + channel_orchestrator.save_config(config) + raise HTTPException(400, f"Failed to enable channel: {e}") + + +@channels.router.post("/{channel_id}/disable") +async def disable_channel(channel_id: str): + config = channel_orchestrator.configs.get(channel_id) + if not config: + raise HTTPException(404, "Channel not found") + + config.enabled = False + config.status = "inactive" + channel_orchestrator.adapters.pop(channel_id, None) + channel_orchestrator.save_config(config) + await ws_events.emit_channel_status(channel_id, "inactive") + return {"ok": True} + + +@channels.router.post("/{channel_id}/test") +async def test_channel(channel_id: str, body: dict | None = None): + config = channel_orchestrator.configs.get(channel_id) + if not config: + raise HTTPException(404, "Channel not found") + + to_number = (body or {}).get("to_number", "") + if not to_number: + raise HTTPException(400, "to_number is required for test") + + try: + adapter = channel_orchestrator.get_adapter(config) + if config.channel_type == "whatsapp": + result = await adapter.send_whatsapp(to_number, config.phone_number, "Test message from Open Swarm") + elif config.channel_type == "voice": + result = {"message": "Voice test: configure webhook and call the number"} + else: + result = await adapter.send_sms(to_number, config.phone_number, "Test message from Open Swarm") + return {"ok": True, "result": result} + except Exception as e: + raise HTTPException(400, f"Test failed: {e}") + + +# ─── Conversations ──────────────────────────────────────────────── + + +@channels.router.get("/{channel_id}/conversations") +async def list_conversations(channel_id: str): + convs = [ + c.model_dump(mode="json") + for c in channel_orchestrator.conversations.values() + if c.channel_id == channel_id + ] + return {"conversations": convs} + + +@channels.router.get("/{channel_id}/conversations/{conversation_id}") +async def get_conversation(channel_id: str, conversation_id: str): + for conv in channel_orchestrator.conversations.values(): + if conv.id == conversation_id and conv.channel_id == channel_id: + return conv.model_dump(mode="json") + raise HTTPException(404, "Conversation not found") + + +# ─── Outbound ───────────────────────────────────────────────────── + + +@channels.router.post("/{channel_id}/send") +async def send_outbound(channel_id: str, body: dict): + to_number = body.get("to_number", "") + message = body.get("message", "") + if not to_number or not message: + raise HTTPException(400, "to_number and message are required") + try: + result = await channel_orchestrator.send_outbound(channel_id, to_number, message) + return result + except ValueError as e: + raise HTTPException(404, str(e)) + + +@channels.router.post("/{channel_id}/call") +async def initiate_call(channel_id: str, body: dict): + to_number = body.get("to_number", "") + if not to_number: + raise HTTPException(400, "to_number is required") + try: + result = await channel_orchestrator.initiate_outbound_call(channel_id, to_number) + return result + except ValueError as e: + raise HTTPException(404, str(e)) + + +# ─── Twilio Webhooks ───────────────────────────────────────────── + + +@channels.router.post("/webhooks/twilio/sms") +async def twilio_sms_webhook(request: Request): + """Inbound SMS webhook from Twilio.""" + form = await request.form() + channel_id = request.query_params.get("channel_id", "") + + # Find channel by phone number if channel_id not provided + if not channel_id: + to_number = form.get("To", "") + for cfg in channel_orchestrator.configs.values(): + if cfg.phone_number == to_number and cfg.channel_type == "sms": + channel_id = cfg.id + break + + config = channel_orchestrator.configs.get(channel_id) + if not config: + return Response(status_code=404) + + # Verify signature + if config.security.verify_signatures: + adapter = channel_orchestrator.get_adapter(config) + sig = request.headers.get("X-Twilio-Signature", "") + url = str(request.url) + if not adapter.verify_webhook_signature(url, dict(form), sig, config.credentials.get("auth_token", "")): + logger.warning("Invalid Twilio signature for channel %s", channel_id) + return Response(status_code=403) + + from_number = form.get("From", "") + body = form.get("Body", "") + num_media = int(form.get("NumMedia", "0")) + media_urls = [form.get(f"MediaUrl{i}", "") for i in range(num_media)] + media_urls = [u for u in media_urls if u] + + await channel_orchestrator.handle_inbound_sms(channel_id, from_number, body, media_urls) + + # Return empty TwiML (Twilio expects XML response) + return Response( + content='', + media_type="application/xml", + ) + + +@channels.router.post("/webhooks/twilio/whatsapp") +async def twilio_whatsapp_webhook(request: Request): + """Inbound WhatsApp webhook from Twilio.""" + form = await request.form() + channel_id = request.query_params.get("channel_id", "") + + if not channel_id: + to_number = form.get("To", "").replace("whatsapp:", "") + for cfg in channel_orchestrator.configs.values(): + if cfg.phone_number == to_number and cfg.channel_type == "whatsapp": + channel_id = cfg.id + break + + config = channel_orchestrator.configs.get(channel_id) + if not config: + return Response(status_code=404) + + if config.security.verify_signatures: + adapter = channel_orchestrator.get_adapter(config) + sig = request.headers.get("X-Twilio-Signature", "") + if not adapter.verify_webhook_signature(str(request.url), dict(form), sig, config.credentials.get("auth_token", "")): + return Response(status_code=403) + + from_number = form.get("From", "").replace("whatsapp:", "") + body = form.get("Body", "") + num_media = int(form.get("NumMedia", "0")) + media_urls = [form.get(f"MediaUrl{i}", "") for i in range(num_media)] + + await channel_orchestrator.handle_inbound_sms(channel_id, from_number, body, media_urls or None) + + return Response( + content='', + media_type="application/xml", + ) + + +@channels.router.post("/webhooks/twilio/voice") +async def twilio_voice_webhook(request: Request): + """Inbound voice call webhook from Twilio.""" + form = await request.form() + channel_id = request.query_params.get("channel_id", "") + + if not channel_id: + to_number = form.get("To", "") + for cfg in channel_orchestrator.configs.values(): + if cfg.phone_number == to_number and cfg.channel_type == "voice": + channel_id = cfg.id + break + + call_sid = form.get("CallSid", "") + from_number = form.get("From", "") + to_number = form.get("To", "") + + twiml = await channel_orchestrator.handle_inbound_call( + channel_id, call_sid, from_number, to_number + ) + + return Response(content=twiml, media_type="application/xml") + + +@channels.router.post("/webhooks/twilio/voice/gather") +async def twilio_voice_gather_webhook(request: Request): + """Speech gathered from a voice call.""" + form = await request.form() + channel_id = request.query_params.get("channel_id", "") + call_sid = request.query_params.get("call_sid", "") or form.get("CallSid", "") + + speech_result = form.get("SpeechResult", "") + + if not speech_result: + # No speech detected, ask again or hang up + config = channel_orchestrator.configs.get(channel_id) + if config: + adapter = channel_orchestrator.get_adapter(config) + voice_cfg = config.voice_config or VoiceConfig() + twiml = adapter.generate_twiml_say( + "I didn't catch that. Goodbye.", voice=voice_cfg.voice + ) + else: + twiml = 'Goodbye.' + return Response(content=twiml, media_type="application/xml") + + twiml = await channel_orchestrator.handle_voice_gather( + channel_id, call_sid, speech_result + ) + + return Response(content=twiml, media_type="application/xml") + + +@channels.router.post("/webhooks/twilio/voice/status") +async def twilio_voice_status_webhook(request: Request): + """Call status update from Twilio.""" + form = await request.form() + call_sid = form.get("CallSid", "") + status = form.get("CallStatus", "") + + channel_orchestrator.handle_call_status(call_sid, status) + return Response(status_code=204) + + +# ─── Telnyx Webhook ─────────────────────────────────────────────── + + +@channels.router.post("/webhooks/telnyx") +async def telnyx_webhook(request: Request): + """Unified Telnyx webhook for SMS and Voice events.""" + body = await request.json() + event_type = body.get("data", {}).get("event_type", "") + payload = body.get("data", {}).get("payload", {}) + + channel_id = request.query_params.get("channel_id", "") + + if event_type == "message.received": + from_number = payload.get("from", {}).get("phone_number", "") + text = payload.get("text", "") + await channel_orchestrator.handle_inbound_sms(channel_id, from_number, text) + elif event_type in ("call.initiated", "call.answered"): + call_sid = payload.get("call_control_id", "") + from_number = payload.get("from", "") + to_number = payload.get("to", "") + # Telnyx voice uses Call Control commands rather than TwiML + logger.info("Telnyx call event: %s for %s", event_type, call_sid) + + return JSONResponse({"ok": True}) + + +# ─── Active Calls ───────────────────────────────────────────────── + + +@channels.router.get("/calls/active") +async def list_active_calls(): + return {"calls": channel_orchestrator.call_manager.get_active_calls()} diff --git a/backend/apps/channels/media_handler.py b/backend/apps/channels/media_handler.py new file mode 100644 index 00000000..71055e4c --- /dev/null +++ b/backend/apps/channels/media_handler.py @@ -0,0 +1,60 @@ +"""Audio attachment processing for WhatsApp voice notes and media messages.""" +import logging +from typing import Optional + +import httpx + +from backend.apps.channels.models import STTConfig +from backend.apps.channels import stt_service + +logger = logging.getLogger(__name__) + +SUPPORTED_FORMATS = { + "audio/ogg", "audio/mpeg", "audio/wav", "audio/mp4", + "audio/flac", "audio/webm", "audio/x-wav", +} +MAX_MEDIA_BYTES = 20 * 1024 * 1024 + + +async def process_audio_attachment( + url: str, + content_type: str, + stt_config: STTConfig, + auth: tuple[str, str] | None = None, +) -> Optional[str]: + """Download an audio attachment and return its transcript. + + Args: + url: URL to download the audio from. + content_type: MIME type of the audio. + stt_config: STT configuration for transcription. + auth: Optional (username, password) tuple for basic auth (e.g. Twilio). + + Returns: + Transcript string, or None on failure. + """ + if content_type not in SUPPORTED_FORMATS: + logger.warning("Unsupported audio format: %s", content_type) + return None + + try: + async with httpx.AsyncClient(timeout=60) as client: + kwargs = {"follow_redirects": True} + if auth: + kwargs["auth"] = auth + resp = await client.get(url, **kwargs) + resp.raise_for_status() + audio_bytes = resp.content + except Exception: + logger.exception("Failed to download audio from %s", url) + return None + + if len(audio_bytes) > MAX_MEDIA_BYTES: + logger.warning("Audio attachment exceeds %d bytes", MAX_MEDIA_BYTES) + return None + + if len(audio_bytes) < 1024: + logger.debug("Audio attachment too small, skipping") + return None + + return await stt_service.transcribe(audio_bytes, stt_config, content_type) diff --git a/backend/apps/channels/models.py b/backend/apps/channels/models.py new file mode 100644 index 00000000..3c0b9502 --- /dev/null +++ b/backend/apps/channels/models.py @@ -0,0 +1,116 @@ +from pydantic import BaseModel, Field +from typing import Optional, Literal, Any +from datetime import datetime +from uuid import uuid4 + + +class ChannelAgentConfig(BaseModel): + mode: str = "agent" + model: str = "sonnet" + system_prompt: Optional[str] = None + max_turns: int = 10 + allowed_tools: Optional[list[str]] = None + + +class ChannelSecurityConfig(BaseModel): + verify_signatures: bool = True + allowlist: list[str] = Field(default_factory=list) + blocklist: list[str] = Field(default_factory=list) + rate_limit_per_minute: int = 10 + rate_limit_per_hour: int = 60 + + +class VoiceConfig(BaseModel): + mode: Literal["conversation", "notify"] = "conversation" + greeting_message: str = "Hello, how can I help you?" + silence_timeout_ms: int = 700 + max_call_duration_seconds: int = 600 + gather_timeout_seconds: int = 10 + voice: str = "Polly.Joanna" + language: str = "en-US" + + +class TTSConfig(BaseModel): + provider: Literal["twilio_say", "elevenlabs", "openai_tts", "edge_tts"] = "twilio_say" + auto_tts_mode: Literal["off", "always", "inbound", "tagged"] = "off" + elevenlabs_voice_id: Optional[str] = None + elevenlabs_model_id: str = "eleven_v3" + openai_voice: str = "alloy" + skip_short_text: bool = True + summarize_long_replies: bool = True + max_tts_chars: int = 4000 + + +class STTConfig(BaseModel): + provider: Literal["twilio_builtin", "deepgram", "openai_whisper"] = "twilio_builtin" + deepgram_model: str = "nova-3" + language: str = "en-US" + fallback_chain: list[str] = Field(default_factory=lambda: ["twilio_builtin"]) + + +class ChannelConfig(BaseModel): + id: str = Field(default_factory=lambda: uuid4().hex) + name: str = "" + channel_type: Literal["sms", "whatsapp", "voice"] = "sms" + provider: Literal["twilio", "telnyx"] = "twilio" + enabled: bool = False + phone_number: str = "" + credentials: dict[str, str] = Field(default_factory=dict) + agent_config: ChannelAgentConfig = Field(default_factory=ChannelAgentConfig) + security: ChannelSecurityConfig = Field(default_factory=ChannelSecurityConfig) + voice_config: Optional[VoiceConfig] = None + tts_config: Optional[TTSConfig] = None + stt_config: Optional[STTConfig] = None + status: Literal["inactive", "active", "error"] = "inactive" + status_message: Optional[str] = None + created_at: str = Field(default_factory=lambda: datetime.now().isoformat()) + updated_at: str = Field(default_factory=lambda: datetime.now().isoformat()) + last_message_at: Optional[str] = None + message_count: int = 0 + + +class ChannelMessage(BaseModel): + id: str = Field(default_factory=lambda: uuid4().hex) + direction: Literal["inbound", "outbound"] = "inbound" + content: str = "" + media_urls: list[str] = Field(default_factory=list) + timestamp: str = Field(default_factory=lambda: datetime.now().isoformat()) + channel_type: str = "" + provider_message_id: Optional[str] = None + + +class ChannelConversation(BaseModel): + id: str = Field(default_factory=lambda: uuid4().hex) + channel_id: str = "" + phone_number: str = "" + agent_session_id: Optional[str] = None + messages: list[ChannelMessage] = Field(default_factory=list) + created_at: str = Field(default_factory=lambda: datetime.now().isoformat()) + updated_at: str = Field(default_factory=lambda: datetime.now().isoformat()) + status: Literal["active", "closed"] = "active" + + +class ChannelCreate(BaseModel): + name: str + channel_type: Literal["sms", "whatsapp", "voice"] = "sms" + provider: Literal["twilio", "telnyx"] = "twilio" + phone_number: str = "" + credentials: dict[str, str] = Field(default_factory=dict) + agent_config: Optional[ChannelAgentConfig] = None + security: Optional[ChannelSecurityConfig] = None + voice_config: Optional[VoiceConfig] = None + tts_config: Optional[TTSConfig] = None + stt_config: Optional[STTConfig] = None + + +class ChannelUpdate(BaseModel): + name: Optional[str] = None + channel_type: Optional[Literal["sms", "whatsapp", "voice"]] = None + provider: Optional[Literal["twilio", "telnyx"]] = None + phone_number: Optional[str] = None + credentials: Optional[dict[str, str]] = None + agent_config: Optional[ChannelAgentConfig] = None + security: Optional[ChannelSecurityConfig] = None + voice_config: Optional[VoiceConfig] = None + tts_config: Optional[TTSConfig] = None + stt_config: Optional[STTConfig] = None diff --git a/backend/apps/channels/orchestrator.py b/backend/apps/channels/orchestrator.py new file mode 100644 index 00000000..4b06979e --- /dev/null +++ b/backend/apps/channels/orchestrator.py @@ -0,0 +1,559 @@ +"""Channel orchestrator — routes inbound messages/calls to agent sessions. + +This is the central routing layer that bridges telephony events to the +existing AgentManager. Each phone number gets its own ChannelConversation +which maps to an AgentSession. +""" +import asyncio +import json +import logging +import os +import time +from datetime import datetime +from typing import Optional + +from backend.apps.channels.models import ( + ChannelConfig, ChannelConversation, ChannelMessage, +) +from backend.apps.channels.call_state import CallManager, CallState +from backend.apps.channels.base_adapter import BaseChannelAdapter +from backend.apps.channels import ws_events +from backend.apps.agents.models import AgentConfig +from backend.config.paths import DATA_ROOT + +logger = logging.getLogger(__name__) + +CHANNELS_DIR = os.path.join(DATA_ROOT, "channels") +CHANNELS_SESSIONS_DIR = os.path.join(DATA_ROOT, "channels", "sessions") + +PLATFORM_MAX_LENGTH = { + "sms": 1600, + "whatsapp": 4096, + "voice": 100000, +} + + +class RateLimiter: + """Simple token-bucket rate limiter per phone number.""" + + def __init__(self): + self._buckets: dict[str, list[float]] = {} + + def check(self, key: str, per_minute: int, per_hour: int) -> bool: + now = time.time() + if key not in self._buckets: + self._buckets[key] = [] + + # Prune old entries + self._buckets[key] = [t for t in self._buckets[key] if now - t < 3600] + + recent_minute = sum(1 for t in self._buckets[key] if now - t < 60) + recent_hour = len(self._buckets[key]) + + if recent_minute >= per_minute or recent_hour >= per_hour: + return False + + self._buckets[key].append(now) + return True + + +class ChannelOrchestrator: + """Manages channel configs, conversations, and message routing.""" + + def __init__(self): + self.configs: dict[str, ChannelConfig] = {} + self.conversations: dict[str, ChannelConversation] = {} # key: "{channel_id}:{phone}" + self.adapters: dict[str, BaseChannelAdapter] = {} + self.call_manager = CallManager() + self.rate_limiter = RateLimiter() + self._agent_listeners: dict[str, asyncio.Task] = {} + + # ─── Config persistence ─────────────────────────────────────── + + def _ensure_dirs(self): + os.makedirs(CHANNELS_DIR, exist_ok=True) + os.makedirs(CHANNELS_SESSIONS_DIR, exist_ok=True) + + def _config_path(self, channel_id: str) -> str: + return os.path.join(CHANNELS_DIR, f"{channel_id}.json") + + def _conv_path(self, channel_id: str) -> str: + return os.path.join(CHANNELS_SESSIONS_DIR, f"{channel_id}.json") + + def save_config(self, config: ChannelConfig): + self._ensure_dirs() + config.updated_at = datetime.now().isoformat() + self.configs[config.id] = config + with open(self._config_path(config.id), "w") as f: + json.dump(config.model_dump(mode="json"), f, indent=2) + + def delete_config(self, channel_id: str): + self.configs.pop(channel_id, None) + self.adapters.pop(channel_id, None) + path = self._config_path(channel_id) + if os.path.exists(path): + os.remove(path) + + def load_all_configs(self): + self._ensure_dirs() + self.configs.clear() + for fname in os.listdir(CHANNELS_DIR): + if fname.endswith(".json"): + try: + with open(os.path.join(CHANNELS_DIR, fname)) as f: + data = json.load(f) + config = ChannelConfig(**data) + self.configs[config.id] = config + except Exception: + logger.exception("Failed to load channel config: %s", fname) + + # ─── Conversation persistence ───────────────────────────────── + + def save_conversations(self, channel_id: str): + self._ensure_dirs() + convs = [ + c.model_dump(mode="json") + for c in self.conversations.values() + if c.channel_id == channel_id + ] + with open(self._conv_path(channel_id), "w") as f: + json.dump(convs, f, indent=2) + + def load_all_conversations(self): + self._ensure_dirs() + self.conversations.clear() + for fname in os.listdir(CHANNELS_SESSIONS_DIR): + if fname.endswith(".json"): + try: + with open(os.path.join(CHANNELS_SESSIONS_DIR, fname)) as f: + convs = json.load(f) + for data in convs: + conv = ChannelConversation(**data) + key = f"{conv.channel_id}:{conv.phone_number}" + self.conversations[key] = conv + except Exception: + logger.exception("Failed to load conversations: %s", fname) + + # ─── Adapter management ─────────────────────────────────────── + + def get_adapter(self, config: ChannelConfig) -> BaseChannelAdapter: + if config.id not in self.adapters: + self.adapters[config.id] = self._create_adapter(config) + return self.adapters[config.id] + + def _create_adapter(self, config: ChannelConfig) -> BaseChannelAdapter: + if config.provider == "twilio": + from backend.apps.channels.adapters.twilio_adapter import TwilioAdapter + return TwilioAdapter( + account_sid=config.credentials.get("account_sid", ""), + auth_token=config.credentials.get("auth_token", ""), + ) + elif config.provider == "telnyx": + from backend.apps.channels.adapters.telnyx_adapter import TelnyxAdapter + return TelnyxAdapter( + api_key=config.credentials.get("api_key", ""), + public_key=config.credentials.get("public_key", ""), + ) + raise ValueError(f"Unknown provider: {config.provider}") + + # ─── Security checks ───────────────────────────────────────── + + def _check_allowlist(self, config: ChannelConfig, phone: str) -> bool: + sec = config.security + if phone in sec.blocklist: + return False + if sec.allowlist and phone not in sec.allowlist: + return False + return True + + def _check_rate_limit(self, config: ChannelConfig, phone: str) -> bool: + sec = config.security + return self.rate_limiter.check( + phone, sec.rate_limit_per_minute, sec.rate_limit_per_hour + ) + + # ─── Inbound SMS / WhatsApp ─────────────────────────────────── + + async def handle_inbound_sms( + self, + channel_id: str, + from_number: str, + body: str, + media_urls: list[str] | None = None, + ) -> Optional[str]: + """Handle an inbound SMS or WhatsApp message. Returns agent response or None.""" + config = self.configs.get(channel_id) + if not config or not config.enabled: + logger.warning("Channel %s not found or disabled", channel_id) + return None + + if not self._check_allowlist(config, from_number): + logger.info("Blocked message from %s (not in allowlist)", from_number) + return None + + if not self._check_rate_limit(config, from_number): + logger.info("Rate limited: %s", from_number) + return None + + # Process media attachments (voice notes) + if media_urls and config.stt_config: + from backend.apps.channels.media_handler import process_audio_attachment + for url in media_urls: + transcript = await process_audio_attachment( + url, "audio/ogg", config.stt_config, + auth=( + config.credentials.get("account_sid", ""), + config.credentials.get("auth_token", ""), + ) if config.provider == "twilio" else None, + ) + if transcript: + body = f"{body}\n\n[Voice Note Transcript]: {transcript}" if body else transcript + + # Get or create conversation + conv_key = f"{channel_id}:{from_number}" + conv = self.conversations.get(conv_key) + if not conv: + conv = ChannelConversation( + channel_id=channel_id, + phone_number=from_number, + ) + self.conversations[conv_key] = conv + + # Record inbound message + inbound_msg = ChannelMessage( + direction="inbound", + content=body, + media_urls=media_urls or [], + channel_type=config.channel_type, + ) + conv.messages.append(inbound_msg) + conv.updated_at = datetime.now().isoformat() + + await ws_events.emit_channel_message( + channel_id, conv.id, inbound_msg.model_dump(mode="json") + ) + + # Launch or reuse agent session + agent_response = await self._route_to_agent(config, conv, body) + + if agent_response: + # Send response back via SMS/WhatsApp + adapter = self.get_adapter(config) + max_len = PLATFORM_MAX_LENGTH.get(config.channel_type, 1600) + chunks = adapter.chunk_message(agent_response, max_len) + + for chunk in chunks: + if config.channel_type == "whatsapp": + await adapter.send_whatsapp(from_number, config.phone_number, chunk) + else: + await adapter.send_sms(from_number, config.phone_number, chunk) + + outbound_msg = ChannelMessage( + direction="outbound", + content=agent_response, + channel_type=config.channel_type, + ) + conv.messages.append(outbound_msg) + conv.updated_at = datetime.now().isoformat() + config.message_count += 1 + config.last_message_at = datetime.now().isoformat() + + await ws_events.emit_channel_message( + channel_id, conv.id, outbound_msg.model_dump(mode="json") + ) + + self.save_conversations(channel_id) + self.save_config(config) + + return agent_response + + # ─── Inbound Voice ──────────────────────────────────────────── + + async def handle_inbound_call( + self, channel_id: str, call_sid: str, from_number: str, to_number: str + ) -> str: + """Handle an inbound voice call. Returns initial TwiML.""" + config = self.configs.get(channel_id) + if not config or not config.enabled: + adapter = self._fallback_adapter(config) + return adapter.generate_twiml_hangup() + + if not self._check_allowlist(config, from_number): + adapter = self.get_adapter(config) + return adapter.generate_twiml_say("Sorry, you are not authorized to call this number.") + + voice_cfg = config.voice_config or VoiceConfig() + adapter = self.get_adapter(config) + + # Create call state + call = self.call_manager.create_call(call_sid, channel_id, from_number, to_number) + call.transition("connected") + call.transition("gathering") + + await ws_events.emit_call_event(channel_id, call_sid, "call_started", { + "from": from_number, "to": to_number, + }) + + # Return TwiML to greet and gather speech + from backend.apps.settings.settings import load_settings + settings = load_settings() + webhook_base = getattr(settings, "webhook_base_url", "") or "" + gather_url = f"{webhook_base}/api/channels/webhooks/twilio/voice/gather?channel_id={channel_id}&call_sid={call_sid}" + + return adapter.generate_twiml_gather( + prompt=voice_cfg.greeting_message, + action_url=gather_url, + voice=voice_cfg.voice, + language=voice_cfg.language, + timeout=voice_cfg.gather_timeout_seconds, + ) + + async def handle_voice_gather( + self, channel_id: str, call_sid: str, speech_result: str + ) -> str: + """Handle gathered speech from a voice call. Returns response TwiML.""" + config = self.configs.get(channel_id) + if not config: + return '' + + call = self.call_manager.get_call(call_sid) + if not call or not call.is_active: + adapter = self.get_adapter(config) + return adapter.generate_twiml_hangup() + + call.transition("processing") + call.add_turn("user", speech_result) + + voice_cfg = config.voice_config or VoiceConfig() + adapter = self.get_adapter(config) + + # Route speech to agent + conv_key = f"{channel_id}:{call.from_number}" + conv = self.conversations.get(conv_key) + if not conv: + conv = ChannelConversation( + channel_id=channel_id, + phone_number=call.from_number, + ) + self.conversations[conv_key] = conv + + agent_response = await self._route_to_agent(config, conv, speech_result) + + if not agent_response: + agent_response = "I'm sorry, I couldn't process that. Could you try again?" + + call.transition("responding") + call.add_turn("assistant", agent_response) + + await ws_events.emit_call_event(channel_id, call_sid, "turn_complete", { + "user": speech_result, "assistant": agent_response, + }) + + # Check if we should continue or end + if voice_cfg.mode == "notify": + call.transition("completed") + return adapter.generate_twiml_say(agent_response, voice=voice_cfg.voice) + + # Conversation mode: say response then gather again + from backend.apps.settings.settings import load_settings + settings = load_settings() + webhook_base = getattr(settings, "webhook_base_url", "") or "" + gather_url = f"{webhook_base}/api/channels/webhooks/twilio/voice/gather?channel_id={channel_id}&call_sid={call_sid}" + + call.transition("gathering") + + return ( + '' + "" + f'{_escape_xml(agent_response)}' + f'' + "" + f'Are you still there? Goodbye.' + "" + ) + + def handle_call_status(self, call_sid: str, status: str): + """Handle Twilio call status callback.""" + call = self.call_manager.get_call(call_sid) + if not call: + return + if status in ("completed", "busy", "no-answer", "canceled", "failed"): + final = "failed" if status == "failed" else "completed" + call.transition(final) + asyncio.create_task( + ws_events.emit_call_event(call.channel_id, call_sid, "call_ended", { + "status": status, + }) + ) + + # ─── Agent routing ──────────────────────────────────────────── + + async def _route_to_agent( + self, config: ChannelConfig, conv: ChannelConversation, text: str + ) -> Optional[str]: + """Send a message to an agent session and wait for the response.""" + from backend.apps.agents.agent_manager import agent_manager + from backend.apps.agents.ws_manager import ws_manager + + # Launch agent if no session exists + if not conv.agent_session_id or not agent_manager.get_session(conv.agent_session_id): + ac = config.agent_config + agent_cfg = AgentConfig( + name=f"{config.channel_type}: {conv.phone_number}", + model=ac.model, + mode=ac.mode, + system_prompt=ac.system_prompt, + max_turns=ac.max_turns, + ) + if ac.allowed_tools: + agent_cfg.allowed_tools = ac.allowed_tools + + session = await agent_manager.launch_agent(agent_cfg) + conv.agent_session_id = session.id + + session_id = conv.agent_session_id + + # Set up a future to capture the agent's response + response_future: asyncio.Future[str] = asyncio.get_event_loop().create_future() + + async def _on_agent_event(event: str, data: dict): + if response_future.done(): + return + if event == "agent:message": + msg = data.get("message", {}) + if msg.get("role") == "assistant": + content = msg.get("content", "") + if isinstance(content, list): + # Extract text from content blocks + parts = [ + b.get("text", "") + for b in content + if isinstance(b, dict) and b.get("type") == "text" + ] + content = "\n".join(parts) + if content and not response_future.done(): + response_future.set_result(content) + elif event == "agent:status": + status = data.get("status", "") + if status in ("completed", "error", "stopped") and not response_future.done(): + response_future.set_result("") + + # Register listener for this session's events + # We tap into ws_manager's send_to_session by monkey-patching temporarily + original_send = ws_manager.send_to_session + + async def _hooked_send(sid: str, event: str, data: dict): + await original_send(sid, event, data) + if sid == session_id: + await _on_agent_event(event, data) + + ws_manager.send_to_session = _hooked_send + + try: + await agent_manager.send_message(session_id, text) + response = await asyncio.wait_for(response_future, timeout=120) + return response if response else None + except asyncio.TimeoutError: + logger.warning("Agent response timed out for session %s", session_id) + return None + except Exception: + logger.exception("Error routing to agent") + return None + finally: + ws_manager.send_to_session = original_send + + def _fallback_adapter(self, config: Optional[ChannelConfig] = None) -> BaseChannelAdapter: + """Return a minimal adapter for generating hangup TwiML.""" + from backend.apps.channels.adapters.twilio_adapter import TwilioAdapter + return TwilioAdapter("", "") + + # ─── Outbound ───────────────────────────────────────────────── + + async def send_outbound( + self, channel_id: str, to_number: str, message: str + ) -> dict: + config = self.configs.get(channel_id) + if not config: + raise ValueError(f"Channel {channel_id} not found") + + adapter = self.get_adapter(config) + max_len = PLATFORM_MAX_LENGTH.get(config.channel_type, 1600) + chunks = adapter.chunk_message(message, max_len) + results = [] + + for chunk in chunks: + if config.channel_type == "whatsapp": + r = await adapter.send_whatsapp(to_number, config.phone_number, chunk) + else: + r = await adapter.send_sms(to_number, config.phone_number, chunk) + results.append(r) + + # Record outbound + conv_key = f"{channel_id}:{to_number}" + conv = self.conversations.get(conv_key) + if not conv: + conv = ChannelConversation(channel_id=channel_id, phone_number=to_number) + self.conversations[conv_key] = conv + + conv.messages.append(ChannelMessage( + direction="outbound", content=message, channel_type=config.channel_type, + )) + conv.updated_at = datetime.now().isoformat() + self.save_conversations(channel_id) + + return {"sent": len(chunks), "results": results} + + async def initiate_outbound_call( + self, channel_id: str, to_number: str + ) -> dict: + config = self.configs.get(channel_id) + if not config: + raise ValueError(f"Channel {channel_id} not found") + + from backend.apps.settings.settings import load_settings + settings = load_settings() + webhook_base = getattr(settings, "webhook_base_url", "") or "" + voice_webhook = f"{webhook_base}/api/channels/webhooks/twilio/voice?channel_id={channel_id}" + + adapter = self.get_adapter(config) + result = await adapter.initiate_call( + to=to_number, + from_=config.phone_number, + webhook_url=voice_webhook, + ) + return result + + # ─── Lifecycle ──────────────────────────────────────────────── + + async def restore_all(self): + self.load_all_configs() + self.load_all_conversations() + for config in self.configs.values(): + if config.enabled: + try: + self.get_adapter(config) + config.status = "active" + except Exception: + config.status = "error" + config.status_message = "Failed to initialize adapter" + + async def persist_all(self): + for config in self.configs.values(): + self.save_config(config) + self.save_conversations(config.id) + + +def _escape_xml(text: str) -> str: + return ( + text.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + .replace("'", "'") + ) + + +# Singleton +from backend.apps.channels.models import VoiceConfig # noqa: E402 +channel_orchestrator = ChannelOrchestrator() diff --git a/backend/apps/channels/stt_service.py b/backend/apps/channels/stt_service.py new file mode 100644 index 00000000..5b9a1aed --- /dev/null +++ b/backend/apps/channels/stt_service.py @@ -0,0 +1,142 @@ +"""Provider-abstracted Speech-to-Text service. + +Supports: Twilio built-in (via Gather), Deepgram Nova-3, OpenAI Whisper. +""" +import logging +from typing import Optional + +import httpx + +from backend.apps.channels.models import STTConfig +from backend.apps.settings.settings import load_settings + +logger = logging.getLogger(__name__) + +SUPPORTED_AUDIO_FORMATS = { + "audio/ogg", "audio/mpeg", "audio/wav", "audio/mp4", + "audio/flac", "audio/webm", "audio/x-wav", +} +MAX_MEDIA_BYTES = 20 * 1024 * 1024 # 20 MB + + +async def transcribe( + audio_bytes: bytes, + config: STTConfig, + content_type: str = "audio/wav", +) -> Optional[str]: + """Transcribe audio bytes to text using the configured provider chain.""" + if len(audio_bytes) > MAX_MEDIA_BYTES: + logger.warning("Audio exceeds %d bytes limit", MAX_MEDIA_BYTES) + return None + if len(audio_bytes) < 1024: + logger.debug("Audio too short, skipping") + return None + + providers = config.fallback_chain or [config.provider] + + for provider in providers: + try: + if provider == "twilio_builtin": + # Twilio STT is handled inline by — no bytes to process + continue + elif provider == "deepgram": + result = await _deepgram_transcribe(audio_bytes, config, content_type) + elif provider == "openai_whisper": + result = await _openai_transcribe(audio_bytes, config, content_type) + else: + logger.warning("Unknown STT provider: %s", provider) + continue + + if result: + return result + except Exception: + logger.exception("STT provider %s failed, trying next", provider) + + return None + + +async def transcribe_from_url( + url: str, config: STTConfig, content_type: str = "audio/ogg" +) -> Optional[str]: + """Download audio from URL and transcribe.""" + try: + async with httpx.AsyncClient(timeout=60) as client: + resp = await client.get(url, follow_redirects=True) + resp.raise_for_status() + return await transcribe(resp.content, config, content_type) + except Exception: + logger.exception("Failed to download audio from %s", url) + return None + + +async def _deepgram_transcribe( + audio_bytes: bytes, config: STTConfig, content_type: str +) -> Optional[str]: + settings = load_settings() + api_key = settings.deepgram_api_key if hasattr(settings, "deepgram_api_key") else None + if not api_key: + logger.warning("Deepgram API key not configured") + return None + + try: + async with httpx.AsyncClient(timeout=60) as client: + resp = await client.post( + "https://api.deepgram.com/v1/listen", + headers={ + "Authorization": f"Token {api_key}", + "Content-Type": content_type, + }, + params={ + "model": config.deepgram_model, + "language": config.language, + "smart_format": "true", + "punctuate": "true", + }, + content=audio_bytes, + ) + resp.raise_for_status() + data = resp.json() + return ( + data.get("results", {}) + .get("channels", [{}])[0] + .get("alternatives", [{}])[0] + .get("transcript", "") + ) + except Exception: + logger.exception("Deepgram transcription failed") + return None + + +async def _openai_transcribe( + audio_bytes: bytes, config: STTConfig, content_type: str +) -> Optional[str]: + settings = load_settings() + api_key = settings.openai_api_key if hasattr(settings, "openai_api_key") else None + if not api_key: + logger.warning("OpenAI API key not configured") + return None + + ext_map = { + "audio/ogg": "ogg", + "audio/mpeg": "mp3", + "audio/wav": "wav", + "audio/x-wav": "wav", + "audio/mp4": "m4a", + "audio/flac": "flac", + "audio/webm": "webm", + } + ext = ext_map.get(content_type, "wav") + + try: + async with httpx.AsyncClient(timeout=60) as client: + resp = await client.post( + "https://api.openai.com/v1/audio/transcriptions", + headers={"Authorization": f"Bearer {api_key}"}, + files={"file": (f"audio.{ext}", audio_bytes, content_type)}, + data={"model": "whisper-1", "language": config.language[:2]}, + ) + resp.raise_for_status() + return resp.json().get("text", "") + except Exception: + logger.exception("OpenAI Whisper transcription failed") + return None diff --git a/backend/apps/channels/talk_mode.py b/backend/apps/channels/talk_mode.py new file mode 100644 index 00000000..c4a0e88f --- /dev/null +++ b/backend/apps/channels/talk_mode.py @@ -0,0 +1,163 @@ +"""Browser-based Talk Mode — continuous voice conversation via WebSocket. + +Pipeline: Mic → WebSocket → STT → Agent → TTS → WebSocket → Speaker + +This runs as a separate WebSocket endpoint /ws/talk/{session_id} that +streams audio bidirectionally between the browser and the STT/TTS services. +""" +import asyncio +import json +import logging +from typing import Optional + +from fastapi import WebSocket, WebSocketDisconnect + +from backend.apps.channels import stt_service, tts_service +from backend.apps.channels.models import STTConfig, TTSConfig +from backend.apps.agents.agent_manager import agent_manager +from backend.apps.agents.ws_manager import ws_manager +from backend.apps.settings.settings import load_settings + +logger = logging.getLogger(__name__) + +# Default configs for talk mode +DEFAULT_STT = STTConfig( + provider="openai_whisper", + fallback_chain=["openai_whisper", "deepgram"], +) +DEFAULT_TTS = TTSConfig( + provider="elevenlabs", + skip_short_text=False, +) + + +async def handle_talk_session(websocket: WebSocket, session_id: str): + """Handle a talk-mode WebSocket connection. + + Protocol: + - Client sends: {"type": "audio", "data": "", "format": "webm"} + - Client sends: {"type": "config", "stt": {...}, "tts": {...}} + - Client sends: {"type": "end_utterance"} when silence detected + - Server sends: {"type": "transcript", "text": "..."} + - Server sends: {"type": "audio", "data": "", "format": "mp3"} + - Server sends: {"type": "agent_response", "text": "..."} + - Server sends: {"type": "status", "status": "listening|processing|speaking"} + """ + await websocket.accept() + + stt_config = DEFAULT_STT + tts_config = DEFAULT_TTS + audio_buffer = bytearray() + + try: + while True: + data = await websocket.receive_text() + msg = json.loads(data) + msg_type = msg.get("type", "") + + if msg_type == "config": + if msg.get("stt"): + stt_config = STTConfig(**msg["stt"]) + if msg.get("tts"): + tts_config = TTSConfig(**msg["tts"]) + await websocket.send_text(json.dumps({"type": "status", "status": "listening"})) + + elif msg_type == "audio": + import base64 + chunk = base64.b64decode(msg.get("data", "")) + audio_buffer.extend(chunk) + + elif msg_type == "end_utterance": + if not audio_buffer: + continue + + await websocket.send_text(json.dumps({"type": "status", "status": "processing"})) + + audio_bytes = bytes(audio_buffer) + audio_buffer.clear() + + audio_format = msg.get("format", "webm") + content_type = f"audio/{audio_format}" + + # STT + transcript = await stt_service.transcribe( + audio_bytes, stt_config, content_type + ) + + if not transcript: + await websocket.send_text(json.dumps({"type": "status", "status": "listening"})) + continue + + await websocket.send_text(json.dumps({ + "type": "transcript", "text": transcript, + })) + + # Route to agent + agent_response = await _get_agent_response(session_id, transcript) + + if agent_response: + await websocket.send_text(json.dumps({ + "type": "agent_response", "text": agent_response, + })) + + # TTS + await websocket.send_text(json.dumps({"type": "status", "status": "speaking"})) + + audio = await tts_service.synthesize(agent_response, tts_config) + if audio: + import base64 as b64 + await websocket.send_text(json.dumps({ + "type": "audio", + "data": b64.b64encode(audio).decode(), + "format": "mp3", + })) + + await websocket.send_text(json.dumps({"type": "status", "status": "listening"})) + + elif msg_type == "stop": + break + + except WebSocketDisconnect: + logger.info("Talk mode disconnected for session %s", session_id) + except Exception: + logger.exception("Talk mode error for session %s", session_id) + + +async def _get_agent_response(session_id: str, text: str) -> Optional[str]: + """Send text to agent and wait for response.""" + session = agent_manager.get_session(session_id) + if not session: + return None + + response_future: asyncio.Future[str] = asyncio.get_event_loop().create_future() + + original_send = ws_manager.send_to_session + + async def _hooked_send(sid: str, event: str, data: dict): + await original_send(sid, event, data) + if sid == session_id and not response_future.done(): + if event == "agent:message": + msg = data.get("message", {}) + if msg.get("role") == "assistant": + content = msg.get("content", "") + if isinstance(content, list): + parts = [ + b.get("text", "") + for b in content + if isinstance(b, dict) and b.get("type") == "text" + ] + content = "\n".join(parts) + if content: + response_future.set_result(content) + elif event == "agent:status": + if data.get("status") in ("completed", "error", "stopped"): + response_future.set_result("") + + ws_manager.send_to_session = _hooked_send + try: + await agent_manager.send_message(session_id, text) + return await asyncio.wait_for(response_future, timeout=120) or None + except asyncio.TimeoutError: + return None + finally: + ws_manager.send_to_session = original_send diff --git a/backend/apps/channels/tts_service.py b/backend/apps/channels/tts_service.py new file mode 100644 index 00000000..c3731724 --- /dev/null +++ b/backend/apps/channels/tts_service.py @@ -0,0 +1,127 @@ +"""Provider-abstracted Text-to-Speech service. + +Supports: Twilio built-in Say, ElevenLabs, OpenAI TTS, Microsoft Edge TTS. +""" +import logging +from typing import Optional + +import httpx + +from backend.apps.channels.models import TTSConfig +from backend.apps.settings.settings import load_settings + +logger = logging.getLogger(__name__) + + +async def synthesize(text: str, config: TTSConfig) -> Optional[bytes]: + """Convert text to audio bytes. Returns None for twilio_say (handled in TwiML).""" + if should_skip(text, config): + return None + + if len(text) > config.max_tts_chars and config.summarize_long_replies: + text = text[: config.max_tts_chars] + + provider = config.provider + if provider == "twilio_say": + # Twilio renders speech inline via — no audio bytes needed + return None + elif provider == "elevenlabs": + return await _elevenlabs_synthesize(text, config) + elif provider == "openai_tts": + return await _openai_synthesize(text, config) + elif provider == "edge_tts": + return await _edge_synthesize(text, config) + + logger.warning("Unknown TTS provider: %s", provider) + return None + + +def should_skip(text: str, config: TTSConfig) -> bool: + if config.skip_short_text and len(text.strip()) < 20: + return True + return False + + +async def _elevenlabs_synthesize(text: str, config: TTSConfig) -> Optional[bytes]: + settings = load_settings() + api_key = settings.elevenlabs_api_key if hasattr(settings, "elevenlabs_api_key") else None + if not api_key: + logger.warning("ElevenLabs API key not configured, falling back to edge_tts") + return await _edge_synthesize(text, config) + + voice_id = config.elevenlabs_voice_id or "21m00Tcm4TlvDq8ikWAM" # Rachel default + model_id = config.elevenlabs_model_id + + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}", + headers={ + "xi-api-key": api_key, + "Content-Type": "application/json", + "Accept": "audio/mpeg", + }, + json={ + "text": text, + "model_id": model_id, + "voice_settings": { + "stability": 0.5, + "similarity_boost": 0.75, + }, + }, + ) + resp.raise_for_status() + return resp.content + except Exception: + logger.exception("ElevenLabs TTS failed, falling back to edge_tts") + return await _edge_synthesize(text, config) + + +async def _openai_synthesize(text: str, config: TTSConfig) -> Optional[bytes]: + settings = load_settings() + api_key = settings.openai_api_key if hasattr(settings, "openai_api_key") else None + if not api_key: + logger.warning("OpenAI API key not configured, falling back to edge_tts") + return await _edge_synthesize(text, config) + + try: + async with httpx.AsyncClient(timeout=30) as client: + resp = await client.post( + "https://api.openai.com/v1/audio/speech", + headers={ + "Authorization": f"Bearer {api_key}", + "Content-Type": "application/json", + }, + json={ + "model": "tts-1", + "input": text, + "voice": config.openai_voice, + "response_format": "mp3", + }, + ) + resp.raise_for_status() + return resp.content + except Exception: + logger.exception("OpenAI TTS failed, falling back to edge_tts") + return await _edge_synthesize(text, config) + + +async def _edge_synthesize(text: str, config: TTSConfig) -> Optional[bytes]: + """Free fallback TTS via Microsoft Edge neural voices. No API key needed.""" + try: + import edge_tts + import tempfile + import os + + communicate = edge_tts.Communicate(text, "en-US-JennyNeural") + with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as f: + tmp_path = f.name + + await communicate.save(tmp_path) + with open(tmp_path, "rb") as f: + audio = f.read() + os.unlink(tmp_path) + return audio + except Exception: + logger.exception("Edge TTS failed") + return None diff --git a/backend/apps/channels/wake_word.py b/backend/apps/channels/wake_word.py new file mode 100644 index 00000000..a79f70cc --- /dev/null +++ b/backend/apps/channels/wake_word.py @@ -0,0 +1,68 @@ +"""Voice Wake Word Detection — scaffolded interface. + +Matches OpenClaw's current state: the interface is defined but full +implementation is deferred. Supports future integration with Vosk +(offline) or Porcupine wake word engines. + +Usage: + This module defines the configuration and interface. Actual wake word + detection runs on the client device (macOS/iOS/Android) and sends + a "wake" event to the gateway when triggered. +""" +import logging +from typing import Optional +from pydantic import BaseModel, Field + +logger = logging.getLogger(__name__) + + +class WakeWordConfig(BaseModel): + """Configuration for wake word detection.""" + enabled: bool = False + wake_words: list[str] = Field(default_factory=lambda: ["hey swarm", "open swarm"]) + sensitivity: float = 0.5 # 0.0 - 1.0 + engine: str = "vosk" # "vosk" | "porcupine" + + +class WakeWordManager: + """Manages wake word detection state. + + In the current scaffolded implementation, this stores configuration + and handles wake events from client devices. The actual audio + processing runs on the client side. + """ + + def __init__(self): + self.config = WakeWordConfig() + self._active_devices: dict[str, bool] = {} + + def update_config(self, **kwargs): + for k, v in kwargs.items(): + if hasattr(self.config, k): + setattr(self.config, k, v) + + def register_device(self, device_id: str): + self._active_devices[device_id] = True + logger.info("Wake word device registered: %s", device_id) + + def unregister_device(self, device_id: str): + self._active_devices.pop(device_id, None) + + def handle_wake_event(self, device_id: str, wake_word: str) -> bool: + """Called when a client device detects a wake word. + + Returns True if the wake event should trigger a talk session. + """ + if not self.config.enabled: + return False + if device_id not in self._active_devices: + return False + if wake_word.lower() not in [w.lower() for w in self.config.wake_words]: + return False + + logger.info("Wake word detected: '%s' from device %s", wake_word, device_id) + return True + + +# Singleton +wake_word_manager = WakeWordManager() diff --git a/backend/apps/channels/ws_events.py b/backend/apps/channels/ws_events.py new file mode 100644 index 00000000..75e0853b --- /dev/null +++ b/backend/apps/channels/ws_events.py @@ -0,0 +1,40 @@ +"""WebSocket event emitters for channel events. + +Uses the existing ws_manager.broadcast_global() — no new WebSocket +infrastructure needed. +""" +import logging +from backend.apps.agents.ws_manager import ws_manager + +logger = logging.getLogger(__name__) + + +async def emit_channel_message( + channel_id: str, conversation_id: str, message: dict +): + await ws_manager.broadcast_global("channel:message", { + "channel_id": channel_id, + "conversation_id": conversation_id, + "message": message, + }) + + +async def emit_channel_status( + channel_id: str, status: str, detail: str = "" +): + await ws_manager.broadcast_global("channel:status", { + "channel_id": channel_id, + "status": status, + "detail": detail, + }) + + +async def emit_call_event( + channel_id: str, call_sid: str, event: str, data: dict | None = None +): + await ws_manager.broadcast_global("channel:call_event", { + "channel_id": channel_id, + "call_sid": call_sid, + "event": event, + **(data or {}), + }) diff --git a/backend/apps/settings/models.py b/backend/apps/settings/models.py index 343b245e..6030e1ab 100644 --- a/backend/apps/settings/models.py +++ b/backend/apps/settings/models.py @@ -21,3 +21,11 @@ class AppSettings(BaseModel): new_agent_shortcut: str = "Meta+l" anthropic_api_key: Optional[str] = None browser_homepage: str = "https://www.google.com" + # Telephony / Channel credentials + twilio_account_sid: Optional[str] = None + twilio_auth_token: Optional[str] = None + telnyx_api_key: Optional[str] = None + elevenlabs_api_key: Optional[str] = None + deepgram_api_key: Optional[str] = None + openai_api_key: Optional[str] = None + webhook_base_url: Optional[str] = None diff --git a/backend/config/paths.py b/backend/config/paths.py index 083ebdfb..1da2d323 100644 --- a/backend/config/paths.py +++ b/backend/config/paths.py @@ -35,5 +35,7 @@ OUTPUTS_WORKSPACE_DIR = os.path.join(DATA_ROOT, "outputs_workspace") SKILLS_WORKSPACE_DIR = os.path.join(DATA_ROOT, "skills_workspace") DASHBOARD_LAYOUT_DIR = os.path.join(DATA_ROOT, "dashboard_layout") BUILTIN_PERMISSIONS_PATH = os.path.join(DATA_ROOT, "builtin_permissions.json") +CHANNELS_DIR = os.path.join(DATA_ROOT, "channels") +CHANNELS_SESSIONS_DIR = os.path.join(DATA_ROOT, "channels", "sessions") BACKEND_DIR = _BACKEND_DIR diff --git a/backend/main.py b/backend/main.py index d7f30cee..4e5b45f1 100644 --- a/backend/main.py +++ b/backend/main.py @@ -16,11 +16,12 @@ from backend.apps.mcp_registry.mcp_registry import mcp_registry from backend.apps.skill_registry.skill_registry import skill_registry from backend.apps.outputs.outputs import outputs from backend.apps.dashboards.dashboards import dashboards +from backend.apps.channels.channels import channels from fastapi.middleware.cors import CORSMiddleware from fastapi import WebSocket, WebSocketDisconnect import json -main_app = MainApp([health, agents, templates, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards]) +main_app = MainApp([health, agents, templates, skills, tools_lib, modes, settings, mcp_registry, skill_registry, outputs, dashboards, channels]) app = main_app.app app.add_middleware( @@ -96,6 +97,12 @@ async def websocket_dashboard(websocket: WebSocket): ws_manager.disconnect_global(websocket) +@app.websocket("/ws/talk/{session_id}") +async def websocket_talk_mode(websocket: WebSocket, session_id: str): + from backend.apps.channels.talk_mode import handle_talk_session + await handle_talk_session(websocket, session_id) + + @app.post("/api/browser/command") async def browser_command(request: Request): """HTTP endpoint called by the browser MCP server subprocess. diff --git a/backend/requirements.txt b/backend/requirements.txt index 8ef474e6..6351bcb0 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -9,4 +9,10 @@ pytest==8.3.4 pytest-asyncio==0.25.2 typeguard==4.4.2 python-dotenv==1.1.1 -Pillow \ No newline at end of file +Pillow +# Channels: SMS, WhatsApp, Voice +twilio>=9.0.0 +telnyx>=2.0.0 +edge-tts>=6.1.0 +httpx>=0.27.0 +cryptography>=42.0.0 \ No newline at end of file diff --git a/frontend/src/app/Main.tsx b/frontend/src/app/Main.tsx index 7628f5f0..c7c3e68b 100644 --- a/frontend/src/app/Main.tsx +++ b/frontend/src/app/Main.tsx @@ -22,6 +22,7 @@ import Tools from './pages/Tools/Tools'; import Modes from './pages/Modes/Modes'; import Views from './pages/Views/Views'; import Customization from './pages/Customization/Customization'; +import Channels from './pages/Channels/Channels'; import { useKeyboardShortcuts } from '@/shared/hooks/useKeyboardShortcuts'; import KeyboardShortcutsHelp from './components/KeyboardShortcutsHelp'; import { ThemeProvider, useThemeMode, useClaudeTokens } from '@/shared/styles/ThemeContext'; @@ -207,6 +208,7 @@ const ThemedApp: React.FC = () => { } /> } /> } /> + } /> diff --git a/frontend/src/app/components/Layout/AppShell.tsx b/frontend/src/app/components/Layout/AppShell.tsx index 2d4edc02..ac9f8740 100644 --- a/frontend/src/app/components/Layout/AppShell.tsx +++ b/frontend/src/app/components/Layout/AppShell.tsx @@ -23,6 +23,7 @@ import ExpandMoreIcon from '@mui/icons-material/ExpandMore'; import AddIcon from '@mui/icons-material/Add'; import SettingsIcon from '@mui/icons-material/Settings'; import ExtensionIcon from '@mui/icons-material/Extension'; +import PhoneIcon from '@mui/icons-material/Phone'; import ViewSidebarOutlinedIcon from '@mui/icons-material/ViewSidebarOutlined'; import ArrowBackOutlinedIcon from '@mui/icons-material/ArrowBackOutlined'; import ArrowForwardOutlinedIcon from '@mui/icons-material/ArrowForwardOutlined'; @@ -129,6 +130,7 @@ const AppShell: React.FC = () => { const isDashboardRoute = location.pathname === '/' || location.pathname.startsWith('/dashboard/'); const isAppsRoute = location.pathname === '/apps' || location.pathname.startsWith('/apps/'); const isCustomizationRoute = location.pathname === '/customization' || CUSTOMIZATION_PATHS.has(location.pathname); + const isChannelsRoute = location.pathname === '/channels'; const activeDashboardId = location.pathname.startsWith('/dashboard/') ? location.pathname.split('/dashboard/')[1] : null; @@ -647,6 +649,38 @@ const AppShell: React.FC = () => { + {/* Divider */} + + + {/* Channels section */} + + navigate('/channels')} + sx={{ + borderRadius: 1.5, + py: 0.6, + px: 1.25, + bgcolor: isChannelsRoute ? `${c.accent.primary}12` : 'transparent', + '&:hover': { bgcolor: isChannelsRoute ? `${c.accent.primary}18` : `${c.text.tertiary}0A` }, + transition: 'background-color 0.15s', + }} + > + + + + + + + {/* Settings */} = { + sms: , + whatsapp: , + voice: , +}; + +const STATUS_COLORS: Record = { + active: '#4caf50', + inactive: '#9e9e9e', + error: '#f44336', +}; + +const Channels: React.FC = () => { + const c = useClaudeTokens(); + const dispatch = useAppDispatch(); + const channels = useAppSelector((s) => s.channels.items); + const conversations = useAppSelector((s) => s.channels.conversations); + const channelList = Object.values(channels).sort( + (a, b) => new Date(b.updated_at).getTime() - new Date(a.updated_at).getTime(), + ); + + const [selectedId, setSelectedId] = useState(null); + const [createOpen, setCreateOpen] = useState(false); + const [tab, setTab] = useState(0); + + // Create form + const [newName, setNewName] = useState(''); + const [newType, setNewType] = useState<'sms' | 'whatsapp' | 'voice'>('sms'); + const [newProvider, setNewProvider] = useState<'twilio' | 'telnyx'>('twilio'); + const [newPhone, setNewPhone] = useState(''); + const [newAccountSid, setNewAccountSid] = useState(''); + const [newAuthToken, setNewAuthToken] = useState(''); + + // Test + const [testNumber, setTestNumber] = useState(''); + const [testResult, setTestResult] = useState(''); + + useEffect(() => { + dispatch(fetchChannels()); + }, [dispatch]); + + const selected = selectedId ? channels[selectedId] : null; + + useEffect(() => { + if (selectedId) dispatch(fetchConversations(selectedId)); + }, [selectedId, dispatch]); + + const handleCreate = async () => { + const result = await dispatch( + createChannel({ + name: newName, + channel_type: newType, + provider: newProvider, + phone_number: newPhone, + credentials: { + account_sid: newAccountSid, + auth_token: newAuthToken, + }, + }), + ); + if (createChannel.fulfilled.match(result)) { + setSelectedId(result.payload.id); + setCreateOpen(false); + setNewName(''); + setNewPhone(''); + setNewAccountSid(''); + setNewAuthToken(''); + } + }; + + const handleDelete = async (id: string) => { + await dispatch(deleteChannel(id)); + if (selectedId === id) setSelectedId(null); + }; + + const handleTest = async () => { + if (!selectedId || !testNumber) return; + try { + const result = await dispatch(testChannel({ id: selectedId, to_number: testNumber })); + if (testChannel.fulfilled.match(result)) { + setTestResult('Test sent successfully!'); + } else { + setTestResult('Test failed'); + } + } catch { + setTestResult('Test failed'); + } + }; + + const convList = Object.values(conversations).filter( + (cv) => cv.channel_id === selectedId, + ); + + return ( + + {/* Left: Channel list */} + + + + Channels + + + setCreateOpen(true)} + sx={{ color: c.accent.primary }} + > + + + + + + + + Connect SMS, WhatsApp, or voice calls to your agents. + + + + + {channelList.length === 0 && ( + + + No channels configured yet + + + )} + {channelList.map((ch) => ( + setSelectedId(ch.id)} + sx={{ + p: 1.5, + mb: 0.5, + borderRadius: 2, + cursor: 'pointer', + bgcolor: selectedId === ch.id ? `${c.accent.primary}14` : 'transparent', + border: selectedId === ch.id ? `1px solid ${c.accent.primary}40` : '1px solid transparent', + '&:hover': { bgcolor: `${c.text.tertiary}0A` }, + transition: 'all 0.15s', + }} + > + + {CHANNEL_TYPE_ICONS[ch.channel_type]} + + + {ch.name} + + + {ch.phone_number} · {ch.provider} + + + + + + ))} + + + + {/* Right: Detail panel */} + + {!selected ? ( + + Select a channel or create a new one + + ) : ( + <> + + {CHANNEL_TYPE_ICONS[selected.channel_type]} + + + {selected.name} + + + + + + + + + {selected.enabled ? ( + + ) : ( + + )} + handleDelete(selected.id)} sx={{ color: c.status.error }}> + + + + + + setTab(v)} sx={{ mb: 2, borderBottom: `1px solid ${c.border.subtle}` }}> + + + + {selected.channel_type === 'voice' && } + + + + + {/* Connection Tab */} + {tab === 0 && ( + + + dispatch(updateChannel({ id: selected.id, phone_number: e.target.value })) + } + /> + + Provider + + + + dispatch( + updateChannel({ + id: selected.id, + credentials: { + ...selected.credentials, + [selected.provider === 'twilio' ? 'account_sid' : 'api_key']: e.target.value, + }, + }), + ) + } + /> + + dispatch( + updateChannel({ + id: selected.id, + credentials: { + ...selected.credentials, + [selected.provider === 'twilio' ? 'auth_token' : 'public_key']: e.target.value, + }, + }), + ) + } + /> + + Webhook URL for Twilio: {window.location.origin.replace(/:\d+$/, ':8324')}/api/channels/webhooks/twilio/{selected.channel_type}?channel_id={selected.id} + + + )} + + {/* Agent Tab */} + {tab === 1 && ( + + + Mode + + + + Model + + + + dispatch( + updateChannel({ + id: selected.id, + agent_config: { ...selected.agent_config, system_prompt: e.target.value || undefined }, + }), + ) + } + /> + + )} + + {/* Security Tab */} + {tab === 2 && ( + + + dispatch( + updateChannel({ + id: selected.id, + security: { ...selected.security, verify_signatures: e.target.checked }, + }), + ) + } + /> + } + label="Verify webhook signatures" + /> + + dispatch( + updateChannel({ + id: selected.id, + security: { + ...selected.security, + allowlist: e.target.value.split('\n').filter(Boolean), + }, + }), + ) + } + /> + + dispatch( + updateChannel({ + id: selected.id, + security: { + ...selected.security, + rate_limit_per_minute: parseInt(e.target.value) || 10, + }, + }), + ) + } + /> + + )} + + {/* Voice Tab */} + {tab === 3 && selected.channel_type === 'voice' && ( + + + Call Mode + + + + dispatch( + updateChannel({ + id: selected.id, + voice_config: { + ...(selected.voice_config || {}), + greeting_message: e.target.value, + } as any, + }), + ) + } + /> + + dispatch( + updateChannel({ + id: selected.id, + voice_config: { + ...(selected.voice_config || {}), + voice: e.target.value, + } as any, + }), + ) + } + /> + + )} + + {/* Conversations Tab */} + {tab === (selected.channel_type === 'voice' ? 4 : 3) && ( + + {convList.length === 0 ? ( + No conversations yet + ) : ( + convList.map((conv) => ( + + + {conv.phone_number} + + + {conv.messages.length} messages · {conv.status} + + + {conv.messages.slice(-5).map((msg) => ( + + + {msg.direction === 'inbound' ? 'Received' : 'Sent'} ·{' '} + {new Date(msg.timestamp).toLocaleTimeString()} + + {msg.content} + + ))} + + + )) + )} + + )} + + {/* Test Tab */} + {tab === (selected.channel_type === 'voice' ? 5 : 4) && ( + + setTestNumber(e.target.value)} + /> + + {testResult && ( + + {testResult} + + )} + + )} + + )} + + + {/* Create Dialog */} + setCreateOpen(false)} maxWidth="sm" fullWidth> + New Channel + + setNewName(e.target.value)} + size="small" + placeholder="My SMS Channel" + /> + + Type + + + + Provider + + + setNewPhone(e.target.value)} + size="small" + placeholder="+1234567890" + /> + setNewAccountSid(e.target.value)} + size="small" + type="password" + /> + setNewAuthToken(e.target.value)} + size="small" + type="password" + /> + + + + + + + + ); +}; + +export default Channels; diff --git a/frontend/src/shared/state/channelsSlice.ts b/frontend/src/shared/state/channelsSlice.ts new file mode 100644 index 00000000..2063cf9e --- /dev/null +++ b/frontend/src/shared/state/channelsSlice.ts @@ -0,0 +1,264 @@ +import { createSlice, createAsyncThunk, PayloadAction } from '@reduxjs/toolkit'; +import { API_BASE } from '@/shared/config'; + +const CHANNELS_API = `${API_BASE}/channels`; + +export interface ChannelAgentConfig { + mode: string; + model: string; + system_prompt?: string; + max_turns: number; + allowed_tools?: string[]; +} + +export interface ChannelSecurityConfig { + verify_signatures: boolean; + allowlist: string[]; + blocklist: string[]; + rate_limit_per_minute: number; + rate_limit_per_hour: number; +} + +export interface VoiceConfig { + mode: 'conversation' | 'notify'; + greeting_message: string; + silence_timeout_ms: number; + max_call_duration_seconds: number; + gather_timeout_seconds: number; + voice: string; + language: string; +} + +export interface TTSConfig { + provider: 'twilio_say' | 'elevenlabs' | 'openai_tts' | 'edge_tts'; + auto_tts_mode: 'off' | 'always' | 'inbound' | 'tagged'; + elevenlabs_voice_id?: string; + elevenlabs_model_id: string; + openai_voice: string; + skip_short_text: boolean; + summarize_long_replies: boolean; + max_tts_chars: number; +} + +export interface STTConfig { + provider: 'twilio_builtin' | 'deepgram' | 'openai_whisper'; + deepgram_model: string; + language: string; + fallback_chain: string[]; +} + +export interface ChannelConfig { + id: string; + name: string; + channel_type: 'sms' | 'whatsapp' | 'voice'; + provider: 'twilio' | 'telnyx'; + enabled: boolean; + phone_number: string; + credentials: Record; + agent_config: ChannelAgentConfig; + security: ChannelSecurityConfig; + voice_config?: VoiceConfig; + tts_config?: TTSConfig; + stt_config?: STTConfig; + status: 'inactive' | 'active' | 'error'; + status_message?: string; + created_at: string; + updated_at: string; + last_message_at?: string; + message_count: number; +} + +export interface ChannelMessage { + id: string; + direction: 'inbound' | 'outbound'; + content: string; + media_urls: string[]; + timestamp: string; + channel_type: string; + provider_message_id?: string; +} + +export interface ChannelConversation { + id: string; + channel_id: string; + phone_number: string; + agent_session_id?: string; + messages: ChannelMessage[]; + created_at: string; + updated_at: string; + status: 'active' | 'closed'; +} + +interface ChannelsState { + items: Record; + conversations: Record; + loading: boolean; + loaded: boolean; +} + +const initialState: ChannelsState = { + items: {}, + conversations: {}, + loading: false, + loaded: false, +}; + +export const fetchChannels = createAsyncThunk( + 'channels/fetch', + async () => { + const res = await fetch(`${CHANNELS_API}/list`); + const data = await res.json(); + return data.channels as ChannelConfig[]; + }, + { condition: (_, { getState }) => !(getState() as { channels: ChannelsState }).channels.loading }, +); + +export const createChannel = createAsyncThunk( + 'channels/create', + async (body: Partial & { name: string }) => { + const res = await fetch(`${CHANNELS_API}/create`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + const data = await res.json(); + return data.channel as ChannelConfig; + }, +); + +export const updateChannel = createAsyncThunk( + 'channels/update', + async ({ id, ...updates }: Partial & { id: string }) => { + const res = await fetch(`${CHANNELS_API}/${id}`, { + method: 'PUT', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(updates), + }); + const data = await res.json(); + return data.channel as ChannelConfig; + }, +); + +export const deleteChannel = createAsyncThunk( + 'channels/delete', + async (id: string) => { + await fetch(`${CHANNELS_API}/${id}`, { method: 'DELETE' }); + return id; + }, +); + +export const enableChannel = createAsyncThunk( + 'channels/enable', + async (id: string) => { + const res = await fetch(`${CHANNELS_API}/${id}/enable`, { method: 'POST' }); + if (!res.ok) throw new Error('Failed to enable'); + return id; + }, +); + +export const disableChannel = createAsyncThunk( + 'channels/disable', + async (id: string) => { + await fetch(`${CHANNELS_API}/${id}/disable`, { method: 'POST' }); + return id; + }, +); + +export const testChannel = createAsyncThunk( + 'channels/test', + async ({ id, to_number }: { id: string; to_number: string }) => { + const res = await fetch(`${CHANNELS_API}/${id}/test`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ to_number }), + }); + if (!res.ok) { + const err = await res.json().catch(() => ({ detail: 'Test failed' })); + throw new Error(err.detail || 'Test failed'); + } + return await res.json(); + }, +); + +export const fetchConversations = createAsyncThunk( + 'channels/fetchConversations', + async (channelId: string) => { + const res = await fetch(`${CHANNELS_API}/${channelId}/conversations`); + const data = await res.json(); + return { channelId, conversations: data.conversations as ChannelConversation[] }; + }, +); + +export const sendOutbound = createAsyncThunk( + 'channels/sendOutbound', + async ({ channelId, to_number, message }: { channelId: string; to_number: string; message: string }) => { + const res = await fetch(`${CHANNELS_API}/${channelId}/send`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ to_number, message }), + }); + return await res.json(); + }, +); + +export const initiateCall = createAsyncThunk( + 'channels/initiateCall', + async ({ channelId, to_number }: { channelId: string; to_number: string }) => { + const res = await fetch(`${CHANNELS_API}/${channelId}/call`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ to_number }), + }); + return await res.json(); + }, +); + +const channelsSlice = createSlice({ + name: 'channels', + initialState, + reducers: { + updateChannelStatus(state, action: PayloadAction<{ channel_id: string; status: string }>) { + const ch = state.items[action.payload.channel_id]; + if (ch) ch.status = action.payload.status as ChannelConfig['status']; + }, + addInboundMessage( + state, + action: PayloadAction<{ channel_id: string; conversation_id: string; message: ChannelMessage }>, + ) { + const conv = state.conversations[action.payload.conversation_id]; + if (conv) { + conv.messages.push(action.payload.message); + } + }, + }, + extraReducers: (builder) => { + builder + .addCase(fetchChannels.pending, (state) => { state.loading = true; }) + .addCase(fetchChannels.fulfilled, (state, action) => { + state.loading = false; + state.loaded = true; + state.items = {}; + for (const c of action.payload) state.items[c.id] = c; + }) + .addCase(fetchChannels.rejected, (state) => { state.loading = false; state.loaded = true; }) + .addCase(createChannel.fulfilled, (state, action) => { state.items[action.payload.id] = action.payload; }) + .addCase(updateChannel.fulfilled, (state, action) => { state.items[action.payload.id] = action.payload; }) + .addCase(deleteChannel.fulfilled, (state, action) => { delete state.items[action.payload]; }) + .addCase(enableChannel.fulfilled, (state, action) => { + const ch = state.items[action.payload]; + if (ch) { ch.enabled = true; ch.status = 'active'; } + }) + .addCase(disableChannel.fulfilled, (state, action) => { + const ch = state.items[action.payload]; + if (ch) { ch.enabled = false; ch.status = 'inactive'; } + }) + .addCase(fetchConversations.fulfilled, (state, action) => { + for (const conv of action.payload.conversations) { + state.conversations[conv.id] = conv; + } + }); + }, +}); + +export const { updateChannelStatus, addInboundMessage } = channelsSlice.actions; +export default channelsSlice.reducer; diff --git a/frontend/src/shared/state/store.ts b/frontend/src/shared/state/store.ts index a1424e45..25bef99a 100644 --- a/frontend/src/shared/state/store.ts +++ b/frontend/src/shared/state/store.ts @@ -12,6 +12,7 @@ import outputsReducer from './outputsSlice'; import dashboardLayoutReducer from './dashboardLayoutSlice'; import dashboardsReducer from './dashboardsSlice'; import updateReducer from './updateSlice'; +import channelsReducer from './channelsSlice'; export const store = configureStore({ reducer: { @@ -28,6 +29,7 @@ export const store = configureStore({ dashboardLayout: dashboardLayoutReducer, dashboards: dashboardsReducer, update: updateReducer, + channels: channelsReducer, }, });