04Connect
Custom endpoints
Your own HTTP and WebSocket routes next to the Realtime API — telephony, webhooks, app protocols.
src/api.py returns Starlette routes. cuelo dev and cuelo serve mount
them before the dashboard, on the same port as /v1/realtime. Cuelo never
overwrites this file.
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import BaseRoute, Route
from cuelo.endpoint.custom import pcm_websocket
from cuelo.runtime.harness import Harness
def routes(harness: Harness) -> list[BaseRoute]:
async def live_calls(request: Request) -> JSONResponse:
return JSONResponse({"calls": sorted(harness.sessions)})
async def say(request: Request) -> JSONResponse:
"""POST /calls/{id}/say {"text": "..."}: speak on a live call, e.g. from a webhook."""
session = harness.sessions.get(request.path_params["id"])
if session is None:
return JSONResponse({"error": "no such live call"}, status_code=404)
body = await request.json()
await session.say(str(body.get("text", "")))
return JSONResponse({"ok": True})
return [
Route("/calls", live_calls),
Route("/calls/{id}/say", say, methods=["POST"]),
pcm_websocket(harness, "/v1/pcm"),
]The session API
Everything an endpoint needs is on the harness and its sessions:
session = await harness.open() # start a call
await session.send_audio(Audio(pcm, 24_000)) # user audio in (PCM16)
async for item in session.output(): # AgentAudio (agent speech) and Event (everything else)
...
await session.say("One moment.") # speak text now, no LLM
await session.respond("typed text") # reply to typed text (or to what was said)
await session.interrupt() # stop the agent
await session.set_voice("voice_id") # switch voice mid-call
await session.close("hangup")
harness.sessions # live sessions by id| Output item | Fields | Notes |
|---|---|---|
AgentAudio | audio, turn_id, backchannel | PCM16 at harness.output_sample_rate (24 kHz). |
Event | kind, t, turn_id, data | Every engine event. On agent.interrupted, drop unplayed audio. |
session.say() doesn't block. It queues the utterance and returns.
The PCM WebSocket
pcm_websocket is a complete endpoint in this style, and a starting point for
your own protocol:
- client → server, binary: user audio, PCM16 mono at 24 kHz.
- client → server, JSON:
{"type": "say", "text"},{"type": "respond", "text"?},{"type": "interrupt"},{"type": "voice", "voice"}. - server → client, binary: agent audio, PCM16 mono at 24 kHz.
- server → client, JSON:
{"type": "session.started", "session_id", "sample_rate"}, then every engine event as{"type": <kind>, "t", "turn_id", "data"}.
Copy cuelo/endpoint/custom.py as the base for Twilio media streams, SIP
gateways or your app's own protocol. Resample and transcode at the edge
(Twilio sends 8 kHz μ-law, for example) and hand the harness PCM16.
Only your routes
To drop the Realtime endpoint and serve only src/api.py:
serve:
api: noneOutside a project
create_app builds the same server around any harness:
from cuelo.serve import create_app
app = create_app(harness, routes=my_routes, api_keys=["..."], cors=["https://app.example.com"])Your routes are protected by the API key like everything else. See Security.