diff --git a/docs/telnyx_media_streaming.md b/docs/telnyx_media_streaming.md new file mode 100644 index 00000000..97736cb3 --- /dev/null +++ b/docs/telnyx_media_streaming.md @@ -0,0 +1,112 @@ +# Telnyx Media Streaming transport (auto-provisioned) + +An alternative to the direct anonymous-WebRTC path for `EVA_FRAMEWORK=telnyx`. Instead of +opening a browser-style WebRTC session (ICE/DTLS/SRTP), EVA places a **Call Control** +outbound call to the assistant's SIP address and bridges audio over a **Media Streaming +WebSocket** (bidirectional PCMU). No WebRTC, so none of the ICE/gateway interop problems of +the WebRTC path apply. + +**You provide only a Telnyx API key.** Everything else — the connection, an idle phone +number for caller ID, the outbound voice profile — is discovered or created for you. + +## 1. Provision (needs only the API key) + +```bash +export TELNYX_API_KEY=KEY... + +# read-only: shows what exists and what is missing, creates nothing +python -m eva.assistant.telnyx_provisioning \ + --assistant-id assistant-473093df-... \ + --public-url https:// + +# create/attach whatever is missing (reuse-first; buys nothing) +python -m eva.assistant.telnyx_provisioning \ + --assistant-id assistant-473093df-... \ + --public-url https:// \ + --provision +``` + +It is **idempotent and reuse-first**: it looks for a Call Control Application named +`eva-media-streaming`, an existing outbound voice profile, and an **unattached** active +number (so it never disturbs the routing of a number already in use). Re-running is a no-op; +if your public URL changed (tunnels do), it re-points the connection's webhook. + +It prints the `EVA_MODEL__S2S_PARAMS` to drop into `.env`: + +```bash +EVA_FRAMEWORK=telnyx +EVA_MODEL__S2S=telnyx +EVA_MODEL__S2S_PARAMS='{"transport":"media_streaming","assistant_id":"assistant-473093df-...","api_key":"KEY...","connection_id":"3004642678097839809","from_number":"+13318719043","to":"sip:anonymous@assistant-473093df-...sip.telnyx.com","webhook_base_url":"https://"}' +``` + +`telnyx_server` selects the media-streaming transport automatically whenever +`connection_id` and `webhook_base_url` are present in `s2s_params` (`_use_call_control`). + +## 2. What "public ingress" means, and why you need it + +The WebRTC path is **outbound-only**: EVA dials out and everything rides that one connection, +so EVA can sit behind NAT with no inbound reachability. **Media Streaming is different — it +requires Telnyx to connect back *to you*.** Two inbound flows: + +1. **Webhooks** — after you start a call, Telnyx sends HTTP POSTs (`call.answered`, + `streaming.started`, …) to `POST {webhook_base_url}/call-control-events`. +2. **The media WebSocket** — Telnyx opens a WebSocket *to your server* at + `wss://{webhook_base_url}/media-stream/{conversation_id}` and streams the call audio over + it, in both directions. + +So Telnyx's servers, out on the public internet, must be able to **reach your EVA process**. +"Public ingress" is exactly that: a publicly resolvable HTTPS/WSS URL that routes inbound +connections to the local port EVA is listening on (default `:PORT` from the run config). + +Your EVA process normally listens on `localhost` (or a private pod IP) that the public +internet cannot reach. You bridge that gap one of two ways: + +- **A tunnel (easiest for a laptop or a locked-down pod).** A tool that opens an outbound + connection to a public relay and gives you back a public URL that forwards to your local + port: + ```bash + # cloudflared (no account needed for a quick tunnel) + cloudflared tunnel --url http://localhost:8080 + # -> https://random-words.trycloudflare.com <- use this as --public-url + + # or ngrok + ngrok http 8080 + # -> https://xxxx.ngrok-free.app + ``` + The tunnel must forward **both** HTTP (webhooks) and WebSocket (media) — cloudflared and + ngrok both do. Note the URL changes each run unless you have a reserved/named tunnel; just + re-run the provisioner to re-point the webhook. + +- **A real ingress (for a deployment).** Expose the EVA service through a Kubernetes + Ingress / load balancer with a public IP and a DNS name and TLS — e.g. a `Service` of + type `LoadBalancer`, or an Ingress resource. Then `--public-url` is that stable hostname. + +Whichever you pick, `webhook_base_url` in `s2s_params` must be that public URL, and it must +terminate TLS (Telnyx requires `https`/`wss`). The provisioner stores it on the connection's +webhook and it is used to build the media-stream URL handed to Telnyx. + +## 3. Run + +```bash +# EVA listens locally; the tunnel/ingress makes it reachable; the call streams audio back. +cloudflared tunnel --url http://localhost:8080 # -> copy the https URL +python -m eva.assistant.telnyx_provisioning --assistant-id ... --public-url --provision +uv run python main.py +``` + +The call flow: EVA `POST /v2/calls` (connection_id, from, to=assistant SIP, `stream_url` = +your `wss://…/media-stream/…`) → Telnyx dials the assistant → `streaming.started` webhook → +Telnyx opens the media WebSocket to you → audio bridges over it, both directions. + +## 4. Why this exists alongside the WebRTC path + +| | WebRTC (`TelnyxWebRTCClient`) | Media Streaming (`TelnyxCallControlTransport`) | +|---|---|---| +| Credentials | public `assistant_id` only | API key (auto-provisions the rest) | +| Inbound reachability | **none needed** (outbound-only) | **public ingress required** | +| Media stack | ICE / DTLS / SRTP (aiortc) | plain WebSocket, PCMU — no ICE | +| Status | blocked on a b2bua-rtc ICE/media interop bug (VSDK-277) | not subject to that bug | + +Use WebRTC when you can only supply an assistant id and can't expose a public URL (e.g. an +upstream EVA maintainer). Use Media Streaming when you have an API key and can provide public +ingress, and want a transport that sidesteps the WebRTC media path entirely. diff --git a/src/eva/assistant/telnyx_provisioning/__init__.py b/src/eva/assistant/telnyx_provisioning/__init__.py new file mode 100644 index 00000000..2c648829 --- /dev/null +++ b/src/eva/assistant/telnyx_provisioning/__init__.py @@ -0,0 +1,18 @@ +"""Telnyx Media Streaming provisioning. + +Given only a Telnyx API key, discover or create every account resource the Call Control +media-streaming transport needs to place a call to an AI Assistant: + + * a Call Control Application (the "connection" — carries the webhook URL and the + outbound-calling settings), + * an owned phone number to use as the caller ID (`from`), attached to that connection, + * an outbound voice profile (so the connection is allowed to place calls). + +Reuse-first and idempotent: nothing is created if a suitable resource already exists, and +re-running is a no-op. Read-only discovery needs no confirmation; creation is gated behind +`provision(create=True)`. +""" + +from .provision import ProvisionResult, TelnyxProvisioner + +__all__ = ["TelnyxProvisioner", "ProvisionResult"] diff --git a/src/eva/assistant/telnyx_provisioning/__main__.py b/src/eva/assistant/telnyx_provisioning/__main__.py new file mode 100644 index 00000000..710b7517 --- /dev/null +++ b/src/eva/assistant/telnyx_provisioning/__main__.py @@ -0,0 +1,4 @@ +from eva.assistant.telnyx_provisioning.provision import main + +if __name__ == "__main__": + main() diff --git a/src/eva/assistant/telnyx_provisioning/provision.py b/src/eva/assistant/telnyx_provisioning/provision.py new file mode 100644 index 00000000..2c40cc37 --- /dev/null +++ b/src/eva/assistant/telnyx_provisioning/provision.py @@ -0,0 +1,255 @@ +"""Idempotent, reuse-first Telnyx account provisioning for the media-streaming transport. + +Usage (read-only discovery — safe, no changes, no confirmation needed): + + python -m eva.assistant.telnyx_provisioning \ + --assistant-id assistant-473093df-... \ + --public-url https://your-tunnel.ngrok-free.app + +Usage (create/attach missing resources — reuses everything it can, buys nothing): + + python -m eva.assistant.telnyx_provisioning \ + --assistant-id assistant-473093df-... \ + --public-url https://your-tunnel.ngrok-free.app \ + --provision + +The API key is read from --api-key or TELNYX_API_KEY. On success it prints the +EVA_MODEL__S2S_PARAMS JSON to drop into your .env. +""" + +from __future__ import annotations + +import argparse +import json +import os +from dataclasses import asdict, dataclass +from typing import Any, Self + +import httpx + +TELNYX_API = "https://api.telnyx.com/v2" + +# Resources we own are tagged/named with this so re-runs find and reuse them. +MANAGED_APP_NAME = "eva-media-streaming" +MANAGED_PROFILE_NAME = "eva-media-streaming" + + +@dataclass +class ProvisionResult: + connection_id: str + from_number: str + to: str + webhook_base_url: str + outbound_voice_profile_id: str | None = None + created: list[str] | None = None + + def s2s_params(self, assistant_id: str, api_key: str) -> dict[str, Any]: + """The EVA_MODEL__S2S_PARAMS payload for the media-streaming transport.""" + return { + "transport": "media_streaming", + "assistant_id": assistant_id, + "api_key": api_key, + "connection_id": self.connection_id, + "from_number": self.from_number, + "to": self.to, + "webhook_base_url": self.webhook_base_url, + } + + +class TelnyxProvisioner: + def __init__(self, api_key: str, *, timeout: float = 30.0) -> None: + if not api_key: + raise ValueError("A Telnyx API key is required (TELNYX_API_KEY or --api-key).") + self._client = httpx.Client( + base_url=TELNYX_API, + headers={"Authorization": f"Bearer {api_key}"}, + timeout=timeout, + ) + self._api_key = api_key + + def close(self) -> None: + self._client.close() + + def __enter__(self) -> Self: + return self + + def __exit__(self, *exc: object) -> None: + self.close() + + # --- low-level helpers --------------------------------------------------- + + def _get(self, path: str, **params: Any) -> list[dict[str, Any]]: + r = self._client.get(path, params=params) + r.raise_for_status() + return r.json().get("data", []) + + def _post(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + r = self._client.post(path, json=payload) + if r.status_code >= 400: + raise RuntimeError(f"POST {path} -> {r.status_code}: {r.text[:400]}") + return r.json().get("data", {}) + + def _patch(self, path: str, payload: dict[str, Any]) -> dict[str, Any]: + r = self._client.patch(path, json=payload) + if r.status_code >= 400: + raise RuntimeError(f"PATCH {path} -> {r.status_code}: {r.text[:400]}") + return r.json().get("data", {}) + + # --- discovery ----------------------------------------------------------- + + def find_connection(self) -> dict[str, Any] | None: + for app in self._get("/call_control_applications", **{"page[size]": 250}): + if app.get("application_name") == MANAGED_APP_NAME: + return app + return None + + def find_outbound_profile(self) -> dict[str, Any] | None: + for prof in self._get("/outbound_voice_profiles", **{"page[size]": 250}): + if prof.get("name") == MANAGED_PROFILE_NAME: + return prof + # fall back to any existing profile so we don't create one unnecessarily + existing = self._get("/outbound_voice_profiles", **{"page[size]": 1}) + return existing[0] if existing else None + + def list_owned_numbers(self) -> list[dict[str, Any]]: + return self._get("/phone_numbers", **{"page[size]": 250, "filter[status]": "active"}) + + def pick_from_number(self, preferred: str | None = None) -> dict[str, Any] | None: + numbers = self.list_owned_numbers() + if preferred: + for n in numbers: + if n.get("phone_number") == preferred: + return n + raise ValueError(f"Requested from-number {preferred} is not active on this account.") + # Prefer a number NOT already attached to a connection, so we never disrupt the + # routing of a number that is in use elsewhere. Among those, prefer US/NANP. + unattached = [n for n in numbers if not n.get("connection_id")] + pool = unattached or numbers + for n in pool: + if str(n.get("phone_number", "")).startswith("+1"): + return n + return pool[0] if pool else None + + # --- provisioning -------------------------------------------------------- + + def ensure( + self, + *, + assistant_id: str, + public_url: str, + from_number: str | None = None, + create: bool = False, + ) -> ProvisionResult: + created: list[str] = [] + webhook_base = public_url.rstrip("/") + + # 1) outbound voice profile (needed for the connection to place calls) + profile = self.find_outbound_profile() + if profile is None: + if not create: + raise _NeedsCreate("outbound_voice_profile", "no outbound voice profile exists") + profile = self._post( + "/outbound_voice_profiles", + {"name": MANAGED_PROFILE_NAME, "traffic_type": "conversational"}, + ) + created.append(f"outbound_voice_profile:{profile['id']}") + profile_id = profile["id"] + + # 2) Call Control Application (the connection), with our webhook URL + conn = self.find_connection() + webhook_event_url = f"{webhook_base}/call-control-events" + if conn is None: + if not create: + raise _NeedsCreate( + "call_control_application", + f"no connection named {MANAGED_APP_NAME!r}", + ) + conn = self._post( + "/call_control_applications", + { + "application_name": MANAGED_APP_NAME, + "webhook_event_url": webhook_event_url, + "outbound": {"outbound_voice_profile_id": profile_id}, + "active": True, + }, + ) + created.append(f"call_control_application:{conn['id']}") + elif conn.get("webhook_event_url") != webhook_event_url: + # keep the webhook pointed at the current public URL (tunnels change) + if create: + conn = self._patch( + f"/call_control_applications/{conn['id']}", + {"webhook_event_url": webhook_event_url}, + ) + created.append(f"call_control_application:{conn['id']}:webhook_updated") + connection_id = conn["id"] + + # 3) a from-number attached to this connection + number = self.pick_from_number(from_number) + if number is None: + raise RuntimeError( + "No active phone number on the account to use as caller ID. " + "Order one in the portal (Numbers -> Buy Numbers) or pass --from-number." + ) + num_id = number["id"] + if number.get("connection_id") != connection_id: + if not create: + raise _NeedsCreate( + "phone_number_assignment", + f"{number['phone_number']} is not attached to the connection", + ) + self._patch(f"/phone_numbers/{num_id}", {"connection_id": connection_id}) + created.append(f"phone_number:{number['phone_number']}:attached") + + # the AI Assistant SIP destination — reachable with just the public assistant id + to = f"sip:anonymous@{assistant_id}.sip.telnyx.com" + + return ProvisionResult( + connection_id=connection_id, + from_number=number["phone_number"], + to=to, + webhook_base_url=webhook_base, + outbound_voice_profile_id=profile_id, + created=created or None, + ) + + +class _NeedsCreate(RuntimeError): + def __init__(self, resource: str, why: str) -> None: + super().__init__(f"missing {resource}: {why} (re-run with --provision to create it)") + self.resource = resource + + +def main() -> None: + ap = argparse.ArgumentParser(description="Provision Telnyx resources for EVA media streaming.") + ap.add_argument("--assistant-id", required=True) + ap.add_argument("--public-url", required=True, help="Public base URL Telnyx can reach (tunnel or ingress).") + ap.add_argument("--api-key", default=os.environ.get("TELNYX_API_KEY")) + ap.add_argument("--from-number", default=None, help="E.164 number to use as caller ID (default: pick one).") + ap.add_argument("--provision", action="store_true", help="Create/attach missing resources (else inspect only).") + args = ap.parse_args() + + with TelnyxProvisioner(args.api_key) as prov: + try: + result = prov.ensure( + assistant_id=args.assistant_id, + public_url=args.public_url, + from_number=args.from_number, + create=args.provision, + ) + except _NeedsCreate as exc: + print(f"[inspect] {exc}") + raise SystemExit(2) from exc + + print("\n=== provisioned ===") + for k, v in asdict(result).items(): + print(f" {k}: {v}") + print("\n=== drop into .env ===") + print("EVA_FRAMEWORK=telnyx") + print("EVA_MODEL__S2S=telnyx") + params = result.s2s_params(args.assistant_id, "") + print(f"EVA_MODEL__S2S_PARAMS='{json.dumps(params)}'") + + +if __name__ == "__main__": + main() diff --git a/tests/unit/assistant/test_telnyx_provisioning.py b/tests/unit/assistant/test_telnyx_provisioning.py new file mode 100644 index 00000000..8a507167 --- /dev/null +++ b/tests/unit/assistant/test_telnyx_provisioning.py @@ -0,0 +1,122 @@ +"""Unit tests for the Telnyx media-streaming provisioner (no network).""" +from __future__ import annotations + +import pytest + +from eva.assistant.telnyx_provisioning.provision import ( + MANAGED_APP_NAME, + ProvisionResult, + TelnyxProvisioner, + _NeedsCreate, +) + + +class _FakeResp: + def __init__(self, data, status=200, text=""): + self._data = data + self.status_code = status + self.text = text or str(data) + + def json(self): + return {"data": self._data} + + def raise_for_status(self): + if self.status_code >= 400: + raise RuntimeError(self.text) + + +class _FakeClient: + def __init__(self, numbers, apps, profiles): + self._numbers = numbers + self._apps = apps + self._profiles = profiles + self.posts: list[tuple[str, dict]] = [] + self.patches: list[tuple[str, dict]] = [] + + def get(self, path, params=None): + if path == "/call_control_applications": + return _FakeResp(self._apps) + if path == "/outbound_voice_profiles": + return _FakeResp(self._profiles) + if path == "/phone_numbers": + return _FakeResp(self._numbers) + return _FakeResp([]) + + def post(self, path, json): + self.posts.append((path, json)) + new = {"id": f"new-{len(self.posts)}", **json} + return _FakeResp(new, status=201) + + def patch(self, path, json): + self.patches.append((path, json)) + return _FakeResp({"id": path.rsplit("/", 1)[-1], **json}) + + def close(self): + pass + + +def _prov(numbers, apps, profiles): + p = TelnyxProvisioner(api_key="test-key") + p._client = _FakeClient(numbers, apps, profiles) + return p + + +def test_requires_api_key(): + with pytest.raises(ValueError): + TelnyxProvisioner(api_key="") + + +def test_inspect_reports_missing_connection(): + prov = _prov( + numbers=[{"id": "n1", "phone_number": "+13125551234", "connection_id": None}], + apps=[], + profiles=[{"id": "prof-1", "name": "existing"}], + ) + with pytest.raises(_NeedsCreate) as exc: + prov.ensure(assistant_id="assistant-abc", public_url="https://x.example", create=False) + assert exc.value.resource == "call_control_application" + + +def test_provision_creates_connection_and_reuses_profile_and_attaches_number(): + prov = _prov( + numbers=[{"id": "n1", "phone_number": "+13125551234", "connection_id": None}], + apps=[], + profiles=[{"id": "prof-1", "name": "existing"}], + ) + result = prov.ensure(assistant_id="assistant-abc", public_url="https://x.example/", create=True) + assert isinstance(result, ProvisionResult) + assert result.from_number == "+13125551234" + assert result.to == "sip:anonymous@assistant-abc.sip.telnyx.com" + assert result.webhook_base_url == "https://x.example" + assert result.outbound_voice_profile_id == "prof-1" # reused, not created + # created the CC app and attached the number + created_paths = [p for p, _ in prov._client.posts] + assert "/call_control_applications" in created_paths + assert prov._client.patches # number attached + + +def test_prefers_unattached_number(): + prov = _prov( + numbers=[ + {"id": "a", "phone_number": "+13120000001", "connection_id": "existing-conn"}, + {"id": "b", "phone_number": "+13120000002", "connection_id": None}, + ], + apps=[{"id": "app-1", "application_name": MANAGED_APP_NAME, + "webhook_event_url": "https://x.example/call-control-events"}], + profiles=[{"id": "prof-1", "name": "existing"}], + ) + result = prov.ensure(assistant_id="assistant-abc", public_url="https://x.example", create=True) + assert result.from_number == "+13120000002" # the unattached one + + +def test_idempotent_when_everything_exists(): + prov = _prov( + numbers=[{"id": "b", "phone_number": "+13120000002", "connection_id": "app-1"}], + apps=[{"id": "app-1", "application_name": MANAGED_APP_NAME, + "webhook_event_url": "https://x.example/call-control-events"}], + profiles=[{"id": "prof-1", "name": "eva-media-streaming"}], + ) + result = prov.ensure(assistant_id="assistant-abc", public_url="https://x.example", create=True) + assert result.created is None # nothing created + assert not prov._client.posts + assert not prov._client.patches