From ce22e94f2a55f5412b73b02eea8bb978d2c1e319 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 12 Sep 2026 19:22:33 +0000 Subject: [PATCH 1/2] feat(vapi): bridge VAPI webhooks into calls and nudges into the live call Co-Authored-By: Aman Ibrahim --- README.md | 49 ++++++ pyproject.toml | 12 +- src/deeptrust/agents/__init__.py | 7 +- src/deeptrust/agents/vapi.py | 258 +++++++++++++++++++++++++++++++ tests/test_adapters.py | 196 +++++++++++++++++++++++ 5 files changed, 519 insertions(+), 3 deletions(-) create mode 100644 src/deeptrust/agents/vapi.py diff --git a/README.md b/README.md index d53e701..8b63b23 100644 --- a/README.md +++ b/README.md @@ -156,6 +156,55 @@ await DeepTrust().watch(conversation_id) # platform="elevenlabs" was already watching. It raises `ServiceError` with status 404 when the platform is not connected for your organisation. +## VAPI + +No extra install: the adapter talks to VAPI over HTTP, and `httpx` is already +here. + +```python +from deeptrust.agents import DeepTrust +from deeptrust.agents.vapi import Bridge + +bridge = Bridge(DeepTrust(), api_key=os.environ["VAPI_API_KEY"]) + +@app.post("/vapi/webhook") # your route, on your server +async def vapi_webhook(payload: dict): + await bridge.handle(payload, user=caller) + return {} +``` + +VAPI is the mirror image of ElevenLabs. Nobody can hold a socket: VAPI posts +its server-url events to *your* server, so the adapter is a handler you call +from your own webhook route rather than a watcher with a loop of its own. Hand +it every event and let it decide — the ones that are not turns cost nothing, +and they carry the call object the control URL is learned from. One bridge +serves every call your server receives. + +Nudges go back on the per-call HTTPS endpoint VAPI publishes as +`monitor.controlUrl`, as an `add-message` with `triggerResponseEnabled: true`. +That is an interrupt, so VAPI behaves like LiveKit rather than ElevenLabs: the +agent responds to the nudge immediately, cutting into what it was saying. It is +sent as a system message, not a `say`, so your agent's own persona carries it +instead of speaking our words verbatim. + +The control URL comes off the webhook payload when the event carries it, and +from `GET /call/{id}` when it does not — which is why the bridge wants a VAPI +private key. Inbound calls are the case this exists for: nobody placed the +call, so there was no creation-time response to capture a URL from. Once +resolved it is remembered for the rest of the call. A call that has already +hung up publishes no control URL, and a nudge from its last turn is dropped +rather than raising inside your webhook route. + +Only final transcripts are read. VAPI emits a `transcript` event per partial +while the sentence is still being recognised, and analysing those would +re-analyse the same sentence several times over. `monitor.listenUrl` next door +is raw PCM audio and is ignored. `end-of-call-report` ends the DeepTrust +session. + +`tool-calls` is the one event whose response controls what the agent does next, +and this adapter does not answer it. Blocking an action is `Session.check`, +which is not implemented in this version. + ## Your own stack Neither adapter is required. If your agent is somewhere else, the two verbs are diff --git a/pyproject.toml b/pyproject.toml index 48f672c..c540130 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -6,7 +6,15 @@ readme = "README.md" requires-python = ">=3.11" license = "Apache-2.0" authors = [{ name = "DeepTrust", email = "engineering@deeptrust.ai" }] -keywords = ["voice", "agents", "livekit", "elevenlabs", "observability", "security"] +keywords = [ + "voice", + "agents", + "livekit", + "elevenlabs", + "vapi", + "observability", + "security", +] classifiers = [ "Development Status :: 3 - Alpha", "Intended Audience :: Developers", @@ -27,6 +35,8 @@ Issues = "https://github.com/deeptrust-ai/deeptrust-python/issues" # integrating their own stack should not be made to install a platform SDK. livekit = ["livekit-agents>=1.7"] elevenlabs = ["websockets>=13"] +# No `vapi` extra: the adapter talks to VAPI over plain HTTP, and httpx is +# already the core dependency. [build-system] requires = ["hatchling"] diff --git a/src/deeptrust/agents/__init__.py b/src/deeptrust/agents/__init__.py index 8957acb..fefa64f 100644 --- a/src/deeptrust/agents/__init__.py +++ b/src/deeptrust/agents/__init__.py @@ -17,8 +17,11 @@ whether a single action may run and does block; it is not implemented in this version. -Adapters for LiveKit and ElevenLabs are in `deeptrust.agents.livekit` and -`deeptrust.agents.elevenlabs`, and wire both ends up for you. +Adapters for LiveKit, ElevenLabs and VAPI are in `deeptrust.agents.livekit`, +`deeptrust.agents.elevenlabs` and `deeptrust.agents.vapi`, and wire both ends +up for you. Which end they hold differs by platform: LiveKit runs in your +process, ElevenLabs gives a socket to hold, and VAPI posts webhooks to your +server and takes nudges back on a per-call control URL. `DeepTrust.watch` is for the hosted path: an organisation that connected its ElevenLabs workspace in the DeepTrust dashboard can hand a live conversation diff --git a/src/deeptrust/agents/vapi.py b/src/deeptrust/agents/vapi.py new file mode 100644 index 0000000..3b51174 --- /dev/null +++ b/src/deeptrust/agents/vapi.py @@ -0,0 +1,258 @@ +"""VAPI adapter. + + from deeptrust.agents import DeepTrust + from deeptrust.agents.vapi import Bridge + + bridge = Bridge(DeepTrust(), api_key=os.environ["VAPI_API_KEY"]) + + @app.post("/vapi/webhook") # your route, your server + async def vapi_webhook(payload: dict): + await bridge.handle(payload, user=caller) + return {} + +VAPI's transport is the mirror image of ElevenLabs'. There is no socket anyone +can hold open: VAPI posts its server-url events to *your* server, and what goes +back the other way goes to a per-call HTTPS endpoint VAPI mints for that call +and publishes on the call object as `monitor.controlUrl`. So the adapter is a +handler you call from inside your own webhook route rather than a watcher with +a loop of its own, and it needs no code inside your agent either way. + +A nudge is delivered as `add-message` with `triggerResponseEnabled: true`, +which is an interrupt: VAPI hands the system message to the model and has it +respond immediately, cutting into what the agent is saying. That makes VAPI +behave like the LiveKit adapter rather than the ElevenLabs one, whose +contextual update is documented as non-interrupting and only shapes the turn +after the current one. A system message rather than a `say` because a `say` +would put our words in the agent's mouth verbatim, while a system message lets +the agent's own persona carry them. + +The control URL is read from the webhook payload when the event carries it, +and fetched with `GET /call/{id}` when it does not, then cached for the rest of +the call. Inbound calls are the case this exists for: nobody placed the call, +so there is no creation-time response to have captured a URL from, and an +adapter that assumed one would work for outbound calls only. + +`monitor.listenUrl` sits next to it and is deliberately ignored: it is a raw +PCM audio stream, not a channel anything can be sent on. + +Only final transcripts are read. VAPI emits a `transcript` event per partial as +the sentence is still being recognised, and analysing those re-analyses the +same sentence several times -- the same class of bug the LiveKit adapter's +`last` dict guards against, arriving here by a different route. + +VAPI's `tool-calls` webhook is the one event whose response controls what the +agent does next, and this adapter does not answer it. Blocking a tool call is +`Session.check`, which is separate work; this is transcript in, nudge out. + +No extra dependency: httpx is already the client's own. +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Any + +import httpx + +from ..errors import ConfigError +from ..types import Analysis, Nudge, User +from . import DeepTrust +from ._session import Session + +API_BASE_URL = "https://api.vapi.ai" + + +def add_message_command(text: str) -> dict[str, Any]: + """The control-URL body that delivers a nudge as an interrupt. + + `triggerResponseEnabled` is what makes it one. Without it VAPI appends the + message and waits for the agent to reach its next turn on its own, which + is a different product: the caller is being worked on now. + """ + return { + "type": "add-message", + "message": {"role": "system", "content": text}, + "triggerResponseEnabled": True, + } + + +class Bridge: + """Turns VAPI server-url events into DeepTrust calls, and nudges into + messages on the live call. + + One bridge serves every call your server receives; state is kept per VAPI + call id, and dropped when the call reports it ended. + """ + + def __init__( + self, + dt: DeepTrust, + *, + api_key: str, + deliver: bool = True, + on_analysis: Callable[[Analysis], None] | None = None, + base_url: str = API_BASE_URL, + ) -> None: + """`api_key` is a VAPI private key: it reads the call object to find + the control URL when an event does not carry one.""" + if not api_key: + raise ConfigError( + "Bridge needs a VAPI private API key. It reads the call to " + "find monitor.controlUrl, which is where a nudge is sent." + ) + self._dt = dt + self._key = api_key + self._deliver = deliver + self._on_analysis = on_analysis + self._base_url = base_url + self._sessions: dict[str, Session] = {} + # Per call, because VAPI mints the URL per call. Cached because most + # events carry it and the fetch is only for the ones that do not. + self._control: dict[str, str] = {} + + async def handle( + self, + payload: dict[str, Any], + *, + user: User | None = None, + ) -> Analysis | None: + """Process one server-url event. Returns the analysis it caused, if any. + + Call it for every event and let it decide: events that are not turns + cost nothing, and the ones that are not transcripts still carry the + call object the control URL is learned from. + + Returns None for an event that started no job, which is most of them. + """ + message = _message(payload) + call = message.get("call") + call = call if isinstance(call, dict) else {} + call_id = str(call.get("id") or "") + if not call_id: + return None + + url = _monitor_control_url(call) + if url: + self._control[call_id] = url + + kind = message.get("type") + if kind == "end-of-call-report": + await self._finish(call_id) + return None + if kind != "transcript": + return None + + role, text = _read_turn(message) + if not text: + return None + + session = self._session(call_id, user) + session.append(role, text) + + # Caller turns only. Feeding the agent's own replies back in doubles + # the work and lets its answers reclassify the call. + if role != "user": + return None + + result = await session.analyze() + if result is None: + return None + if self._on_analysis: + self._on_analysis(result) + if self._deliver: + for nudge in result.nudges: + await self.send_nudge(call_id, nudge) + return result + + async def send_nudge(self, call_id: str, nudge: Nudge) -> bool: + """Send one nudge into a live call. Returns whether VAPI took it. + + A call that has already ended publishes no control URL, so a nudge + produced from its last turn returns False rather than raising: the + call it was for is over, and the finding is already recorded. + """ + url = await self.control_url(call_id) + if not url: + return False + + async with httpx.AsyncClient() as client: + # No credential on this request. The control URL carries its own + # authority and VAPI does not accept the private key here. + response = await client.post(url, json=add_message_command(nudge.render())) + return response.is_success + + async def control_url(self, call_id: str) -> str | None: + """The call's `monitor.controlUrl`, from cache or from VAPI.""" + cached = self._control.get(call_id) + if cached: + return cached + + async with httpx.AsyncClient( + base_url=self._base_url, + headers={"Authorization": f"Bearer {self._key}"}, + ) as client: + response = await client.get(f"/call/{call_id}") + if not response.is_success: + return None + + body = response.json() + url = _monitor_control_url(body if isinstance(body, dict) else {}) + if url: + self._control[call_id] = url + return url + + def session(self, call_id: str) -> Session | None: + """The DeepTrust session for a call, so its transcript stays reachable.""" + return self._sessions.get(call_id) + + def _session(self, call_id: str, user: User | None) -> Session: + session = self._sessions.get(call_id) + if session is None: + session = self._dt.session(external_id=call_id, user=user, platform="vapi") + self._sessions[call_id] = session + return session + + async def _finish(self, call_id: str) -> None: + self._control.pop(call_id, None) + session = self._sessions.pop(call_id, None) + if session is not None: + await session.end() + + +def _message(payload: dict[str, Any]) -> dict[str, Any]: + """The event itself, out of the request body. + + VAPI wraps a server-url event in `{"message": {...}}`. A bare event is + accepted too, so a payload already unwrapped by the caller's own framework + still works. + """ + message = payload.get("message") + return message if isinstance(message, dict) else payload + + +def _read_turn(message: dict[str, Any]) -> tuple[str, str]: + """A turn from a transcript event, or ("", "") if it is not one yet. + + Partials are not turns. VAPI sends one event per revision of the sentence + being recognised, all with the same `transcriptType: "partial"`, and only + the final one is the sentence the caller actually said. + """ + if message.get("transcriptType") != "final": + return "", "" + role = "user" if str(message.get("role") or "") == "user" else "agent" + return role, str(message.get("transcript") or "").strip() + + +def _monitor_control_url(call: dict[str, Any]) -> str | None: + """`monitor.controlUrl` off a call object, or None. + + Defensive about the shape rather than trusting it: this runs on the + webhook path, and a missing or renamed field has to read as "no control + URL yet" -- which a fetch may still answer -- instead of as a TypeError + inside the customer's webhook route. + """ + monitor = call.get("monitor") + if not isinstance(monitor, dict): + return None + url = monitor.get("controlUrl") + return url if isinstance(url, str) and url else None diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 6715c11..3adf102 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -6,6 +6,7 @@ from __future__ import annotations +import json from typing import Any import httpx @@ -14,6 +15,8 @@ from deeptrust.agents import DeepTrust from deeptrust.agents.elevenlabs import Monitor, _read_turn, contextual_update_command from deeptrust.agents.livekit import attach +from deeptrust.agents.vapi import Bridge, add_message_command +from deeptrust.agents.vapi import _read_turn as _vapi_read_turn BASE = "https://example.test/api/v1" @@ -244,3 +247,196 @@ async def test_elevenlabs_monitor_sends_nudges_in_the_command_envelope() -> None }, } ] + + +def test_vapi_add_message_is_an_interrupting_system_message() -> None: + """`triggerResponseEnabled` is the whole difference between a nudge that + cuts in and one that waits for the agent's next turn.""" + assert add_message_command("hold the line") == { + "type": "add-message", + "message": {"role": "system", "content": "hold the line"}, + "triggerResponseEnabled": True, + } + + +def test_vapi_reads_final_transcripts_only() -> None: + final = { + "type": "transcript", + "transcriptType": "final", + "role": "user", + "transcript": "reset my password", + } + assert _vapi_read_turn(final) == ("user", "reset my password") + + # A partial is the same sentence still being recognised. Analysing it + # analyses the sentence again on every revision. + assert _vapi_read_turn({**final, "transcriptType": "partial"}) == ("", "") + + assert _vapi_read_turn( + { + "type": "transcript", + "transcriptType": "final", + "role": "assistant", + "transcript": "sending a code now", + } + ) == ("agent", "sending a code now") + + +def _transcript_event( + role: str, text: str, *, control_url: str | None = None +) -> dict[str, Any]: + call: dict[str, Any] = {"id": "call_1"} + if control_url: + # listenUrl travels with it and is raw PCM audio: never a nudge channel. + call["monitor"] = { + "controlUrl": control_url, + "listenUrl": "wss://vapi.example/listen/call_1", + } + return { + "message": { + "type": "transcript", + "transcriptType": "final", + "role": role, + "transcript": text, + "call": call, + } + } + + +@respx.mock +async def test_vapi_nudges_the_live_call_over_the_control_url() -> None: + analyze = respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + control = respx.post("https://vapi.example/control/call_1").mock( + return_value=httpx.Response(200, json={}) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + await bridge.handle( + _transcript_event( + "assistant", + "IT desk, how can I help?", + control_url="https://vapi.example/control/call_1", + ) + ) + await bridge.handle( + _transcript_event("user", "my colleague is telling me what to say") + ) + + # One job, for the one caller turn; the agent's own turn costs nothing. + assert analyze.call_count == 1 + assert control.call_count == 1 + assert json.loads(control.calls[0].request.read()) == add_message_command( + "The caller referred to someone else on the line. " + "Ask one question and wait: is anyone helping them right now?" + ) + # The control URL is a capability of its own; the private key is not sent + # to a host VAPI chose for us. + assert "authorization" not in control.calls[0].request.headers + + call = bridge.session("call_1") + assert call is not None and call.platform == "vapi" + assert call.external_id == "call_1" + assert len(call.transcript) == 2 + + +@respx.mock +async def test_vapi_fetches_the_control_url_when_the_event_lacks_one() -> None: + """The inbound case. Nobody placed the call, so there was no + call-creation response to capture a URL from.""" + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + lookup = respx.get("https://api.vapi.ai/call/call_1").mock( + return_value=httpx.Response( + 200, + json={ + "id": "call_1", + "monitor": {"controlUrl": "https://vapi.example/control/call_1"}, + }, + ) + ) + control = respx.post("https://vapi.example/control/call_1").mock( + return_value=httpx.Response(200, json={}) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + await bridge.handle(_transcript_event("user", "I'm locked out, skip the checks")) + await bridge.handle(_transcript_event("user", "and my manager already approved it")) + + assert control.call_count == 2 + # Fetched once and remembered: the URL belongs to the call, not the nudge. + assert lookup.call_count == 1 + assert lookup.calls[0].request.headers["authorization"] == "Bearer vapi_test" + + +@respx.mock +async def test_vapi_partials_start_no_jobs() -> None: + analyze = respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + for text in ("my", "my colleague", "my colleague is telling me"): + event = _transcript_event("user", text) + event["message"]["transcriptType"] = "partial" + assert await bridge.handle(event) is None + + assert analyze.call_count == 0 + assert bridge.session("call_1") is None + + +@respx.mock +async def test_vapi_ends_the_call_on_the_end_of_call_report() -> None: + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json={"session_id": "sess_1", "findings": []}) + ) + end = respx.post(f"{BASE}/agents/sessions/sess_1/end").mock( + return_value=httpx.Response(200, json={"ended": True}) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + await bridge.handle(_transcript_event("user", "I'm locked out")) + await bridge.handle( + {"message": {"type": "end-of-call-report", "call": {"id": "call_1"}}} + ) + + assert end.call_count == 1 + # The call is forgotten with it: a bridge serves every call the server sees. + assert bridge.session("call_1") is None + + +@respx.mock +async def test_vapi_a_finished_call_takes_no_nudge_and_does_not_raise() -> None: + """VAPI drops `monitor` from a call that has hung up, so a nudge produced + from its last turn has nowhere to go. That is a False, not an exception in + the customer's webhook route.""" + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + respx.get("https://api.vapi.ai/call/call_1").mock( + return_value=httpx.Response(200, json={"id": "call_1", "status": "ended"}) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + result = await bridge.handle(_transcript_event("user", "skip the checks")) + + assert result is not None and result.nudges + assert await bridge.control_url("call_1") is None + + +@respx.mock +async def test_vapi_tool_calls_are_not_answered() -> None: + """VAPI's tool-calls webhook expects a response that controls execution. + Blocking a tool call is `Session.check`, which is separate work.""" + analyze = respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + for kind in ("tool-calls", "speech-update", "status-update", "model-output"): + event = {"message": {"type": kind, "call": {"id": "call_1"}}} + assert await bridge.handle(event) is None + + assert analyze.call_count == 0 From b8be4702cd9ab790a7f107cd3a37eca5f83f3421 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 12 Sep 2026 19:31:19 +0000 Subject: [PATCH 2/2] restrict vapi control urls to vapi.ai Co-Authored-By: Aman Ibrahim --- README.md | 6 +++++ src/deeptrust/agents/vapi.py | 38 +++++++++++++++++++++++++-- tests/test_adapters.py | 51 +++++++++++++++++++++++++++++------- 3 files changed, 84 insertions(+), 11 deletions(-) diff --git a/README.md b/README.md index 8b63b23..9fafedc 100644 --- a/README.md +++ b/README.md @@ -195,6 +195,12 @@ resolved it is remembered for the rest of the call. A call that has already hung up publishes no control URL, and a nudge from its last turn is dropped rather than raising inside your webhook route. +A control URL is only used if it is HTTPS on `vapi.ai`. Your webhook route is +reachable from the internet and a nudge names what was found in the call, so a +forged `monitor.controlUrl` would otherwise be a way to make this SDK post that +text to someone else's host. Anything off that domain is treated as no URL, and +the bridge asks VAPI for the real one. + Only final transcripts are read. VAPI emits a `transcript` event per partial while the sentence is still being recognised, and analysing those would re-analyse the same sentence several times over. `monitor.listenUrl` next door diff --git a/src/deeptrust/agents/vapi.py b/src/deeptrust/agents/vapi.py index 3b51174..1fba757 100644 --- a/src/deeptrust/agents/vapi.py +++ b/src/deeptrust/agents/vapi.py @@ -35,6 +35,13 @@ async def vapi_webhook(payload: dict): `monitor.listenUrl` sits next to it and is deliberately ignored: it is a raw PCM audio stream, not a channel anything can be sent on. +A control URL is only accepted if it is HTTPS on VAPI's own domain. The webhook +body is attacker-reachable in the general case -- it arrives over the public +internet at your route -- and a nudge names what DeepTrust found in the call, so +a forged `monitor.controlUrl` would be a way to have this SDK post that text to +a host of someone else's choosing. Anything off `vapi.ai` reads as no control +URL rather than as an error. + Only final transcripts are read. VAPI emits a `transcript` event per partial as the sentence is still being recognised, and analysing those re-analyses the same sentence several times -- the same class of bug the LiveKit adapter's @@ -51,6 +58,7 @@ async def vapi_webhook(payload: dict): from collections.abc import Callable from typing import Any +from urllib.parse import quote, urlsplit import httpx @@ -61,6 +69,9 @@ async def vapi_webhook(payload: dict): API_BASE_URL = "https://api.vapi.ai" +#: The only domain a control URL may point at. +CONTROL_URL_DOMAIN = "vapi.ai" + def add_message_command(text: str) -> dict[str, Any]: """The control-URL body that delivers a nudge as an interrupt. @@ -191,7 +202,10 @@ async def control_url(self, call_id: str) -> str | None: base_url=self._base_url, headers={"Authorization": f"Bearer {self._key}"}, ) as client: - response = await client.get(f"/call/{call_id}") + # Quoted: the id comes off a webhook body, and a raw `/` or `?` in + # it would address a different endpoint of the API than the call + # lookup. + response = await client.get(f"/call/{quote(call_id, safe='')}") if not response.is_success: return None @@ -255,4 +269,24 @@ def _monitor_control_url(call: dict[str, Any]) -> str | None: if not isinstance(monitor, dict): return None url = monitor.get("controlUrl") - return url if isinstance(url, str) and url else None + if not isinstance(url, str) or not url: + return None + return url if _is_vapi_control_url(url) else None + + +def _is_vapi_control_url(url: str) -> bool: + """Whether a URL is one VAPI could have minted: HTTPS, on `vapi.ai`. + + The check is on the host rather than the full URL because VAPI mints these + per region and per call -- the path and the subdomain both vary -- while + the domain is the part that says the destination is VAPI and not somewhere + a forged webhook pointed us. + """ + try: + parsed = urlsplit(url) + except ValueError: + return False + if parsed.scheme != "https": + return False + host = (parsed.hostname or "").lower() + return host == CONTROL_URL_DOMAIN or host.endswith(f".{CONTROL_URL_DOMAIN}") diff --git a/tests/test_adapters.py b/tests/test_adapters.py index 3adf102..ba28aae 100644 --- a/tests/test_adapters.py +++ b/tests/test_adapters.py @@ -19,6 +19,11 @@ from deeptrust.agents.vapi import _read_turn as _vapi_read_turn BASE = "https://example.test/api/v1" +# Shaped like the real thing: VAPI mints these per region and per call, and only +# the domain is fixed. +CONTROL_URL = ( + "https://aws-us-west-2-production1-phone-call-websocket.vapi.ai/call_1/control" +) ONE_NUDGE = { "session_id": "sess_1", @@ -290,7 +295,10 @@ def _transcript_event( # listenUrl travels with it and is raw PCM audio: never a nudge channel. call["monitor"] = { "controlUrl": control_url, - "listenUrl": "wss://vapi.example/listen/call_1", + "listenUrl": ( + "wss://aws-us-west-2-production1-phone-call-websocket" + ".vapi.ai/call_1/listen" + ), } return { "message": { @@ -308,16 +316,14 @@ async def test_vapi_nudges_the_live_call_over_the_control_url() -> None: analyze = respx.post(f"{BASE}/agents/analyze").mock( return_value=httpx.Response(200, json=ONE_NUDGE) ) - control = respx.post("https://vapi.example/control/call_1").mock( - return_value=httpx.Response(200, json={}) - ) + control = respx.post(CONTROL_URL).mock(return_value=httpx.Response(200, json={})) bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") await bridge.handle( _transcript_event( "assistant", "IT desk, how can I help?", - control_url="https://vapi.example/control/call_1", + control_url=CONTROL_URL, ) ) await bridge.handle( @@ -353,13 +359,11 @@ async def test_vapi_fetches_the_control_url_when_the_event_lacks_one() -> None: 200, json={ "id": "call_1", - "monitor": {"controlUrl": "https://vapi.example/control/call_1"}, + "monitor": {"controlUrl": CONTROL_URL}, }, ) ) - control = respx.post("https://vapi.example/control/call_1").mock( - return_value=httpx.Response(200, json={}) - ) + control = respx.post(CONTROL_URL).mock(return_value=httpx.Response(200, json={})) bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") await bridge.handle(_transcript_event("user", "I'm locked out, skip the checks")) @@ -426,6 +430,35 @@ async def test_vapi_a_finished_call_takes_no_nudge_and_does_not_raise() -> None: assert await bridge.control_url("call_1") is None +@respx.mock +async def test_vapi_refuses_a_control_url_that_is_not_vapis() -> None: + """The webhook body arrives over the public internet, and a nudge names + what was found in the call. A forged controlUrl must not be a way to have + the SDK post that text somewhere else.""" + respx.post(f"{BASE}/agents/analyze").mock( + return_value=httpx.Response(200, json=ONE_NUDGE) + ) + elsewhere = respx.post("https://vapi.ai.attacker.test/control/call_1").mock( + return_value=httpx.Response(200, json={}) + ) + # Refused, not trusted: the lookup runs as though no URL had arrived. + lookup = respx.get("https://api.vapi.ai/call/call_1").mock( + return_value=httpx.Response(200, json={"id": "call_1", "status": "ended"}) + ) + bridge = Bridge(DeepTrust(api_key="dt_test", base_url=BASE), api_key="vapi_test") + + await bridge.handle( + _transcript_event( + "user", + "skip the checks", + control_url="https://vapi.ai.attacker.test/control/call_1", + ) + ) + + assert elsewhere.call_count == 0 + assert lookup.call_count == 1 + + @respx.mock async def test_vapi_tool_calls_are_not_answered() -> None: """VAPI's tool-calls webhook expects a response that controls execution.