# Custom endpoints (/connect/custom-endpoints)

`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.

```python title="src/api.py"
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 [#the-session-api]

Everything an endpoint needs is on the harness and its sessions:

```python
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 [#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 [#only-your-routes]

To drop the Realtime endpoint and serve only `src/api.py`:

```yaml title="config.cuelo"
serve:
  api: none
```

## Outside a project [#outside-a-project]

`create_app` builds the same server around any harness:

```python
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](/ship/security).