# Add a provider (/contributing/providers)

Providers are the most common contribution. Adding one touches six places,
and the tests catch you if you miss one:

| #   | Where                                                   | What                                                   |
| --- | ------------------------------------------------------- | ------------------------------------------------------ |
| 1   | `engine/src/cuelo/{stt,tts,llms}/<vendor>.py`           | the class, implementing the slot interface             |
| 2   | `engine/pyproject.toml`                                 | an optional-dependency extra, if it needs packages     |
| 3   | `schema/config.cuelo.schema.json`                       | the name in the provider `enum`, plus a registry entry |
| 4   | `make schema`                                           | copies the schema into the engine and CLI              |
| 5   | `cli/internal/cli/init.go` (`orderedNames`)             | so `cuelo init --stt/--tts/--llm` lists it             |
| 6   | `engine/tests/`, `docs/content/reference/providers.mdx` | tests against a fake server, and a docs row            |

TTS providers can also get a voice catalogue for `cuelo voices` (step 7).

<RegistryDiagram />

We'll walk through each step with a speech-to-text provider for a made-up
vendor, **Acme**, then cover what's different for TTS and LLMs.

<Callout title="Before you start">
Read the vendor's streaming docs end to end: auth, URL and query parameters, message shapes, how
to force a final transcript, how the session closes, and error messages. Link them in the module
docstring. Every provider in the repo does, and reviewers check the code against them.
</Callout>

<Steps>
<Step>
### Write the class [#write-the-class]

Every provider subclasses its slot interface, which subclasses
`cuelo.provider.Provider`. The base class resolves `base_url` and the API key
for you:

- `default_base_url`: the endpoint when `base_url` isn't set.
- `default_api_key_env`: the variable read when `api_key` and `api_key_env`
  aren't given. `self.api_key` returns the key or raises
  `MissingCredentialError` with a helpful message.
- `supports`: which optional settings the provider honors. Setting any other
  one raises at construction, so a config that asks for something unsupported
  fails loudly instead of being silently ignored.

An STT provider implements `connect()` and returns an `STTSession` with four
methods. Here is the whole thing for a WebSocket API:

```python title="engine/src/cuelo/stt/acme.py"
"""Acme real-time speech-to-text.

The first message is a JSON config; audio follows as binary PCM frames. The
server sends {"type": "partial" | "final", "text", "start", "end"}.
`finalize()` sends {"type": "flush"}; the server answers with a final that
has "flushed": true, which marks the end of the turn.

`language` becomes `lang`; `vocab` becomes `boost`.

Docs: https://docs.acme.example/stt/streaming
"""

import json
from collections.abc import AsyncIterator
from typing import Any

from websockets.asyncio.client import ClientConnection, connect

from cuelo.provider import ProviderError, primary_language, ws_url
from cuelo.runtime.audio import SAMPLE_WIDTH, Audio, PCMBuffer
from cuelo.stt.interface import STT, STTSession, Transcript

CHUNK_SECONDS = 0.1

class AcmeSTT(STT):
    default_base_url = "https://stream.acme.example"
    default_api_key_env = "ACME_API_KEY"
    supports = frozenset({"language", "vocab"})

    def __init__(self, *, model: str = "acme-rt-2", **kwargs: Any) -> None:
        super().__init__(**kwargs)   # language, vocab, sample_rate, base_url, api_key, api_key_env
        self.model = model

    def config(self) -> dict[str, Any]:
        config: dict[str, Any] = {
            "model": self.model,
            "encoding": "pcm_s16le",
            "sample_rate": self.sample_rate,
        }
        if self.language:
            config["lang"] = primary_language(self.language)   # "en-US" -> "en"
        if self.vocab:
            config["boost"] = self.vocab
        return config

    async def connect(self) -> STTSession:
        ws = await connect(
            f"{ws_url(self.base_url or '')}/v2/listen",
            additional_headers={"Authorization": f"Bearer {self.api_key}"},
        )
        await ws.send(json.dumps(self.config()))
        return _Session(ws, self.sample_rate)

class _Session(STTSession):
    def __init__(self, ws: ClientConnection, sample_rate: int) -> None:
        self.ws = ws
        self.sample_rate = sample_rate
        # Batch the harness's small frames into ~100 ms chunks.
        self.pcm = PCMBuffer(min_bytes=int(sample_rate * CHUNK_SECONDS) * SAMPLE_WIDTH)

    async def send(self, audio: Audio) -> None:
        if audio.sample_rate != self.sample_rate:
            raise ValueError(f"expected {self.sample_rate} Hz audio, got {audio.sample_rate}")
        if data := self.pcm.push(audio.pcm):
            await self.ws.send(data)

    async def finalize(self) -> None:
        if data := self.pcm.flush():
            await self.ws.send(data)
        await self.ws.send(json.dumps({"type": "flush"}))

    async def transcripts(self) -> AsyncIterator[Transcript]:
        async for raw in self.ws:
            msg = json.loads(raw)
            if msg.get("type") == "error":
                raise ProviderError(f"Acme STT: {msg.get('message')}")
            if msg.get("type") in ("partial", "final") and msg.get("text"):
                yield Transcript(
                    msg["text"],
                    is_final=msg["type"] == "final",
                    finalized=bool(msg.get("flushed")),
                    start=msg.get("start"),
                    end=msg.get("end"),
                )

    async def close(self) -> None:
        await self.ws.close()
```

What the harness relies on:

- **`sample_rate`** is what you want to receive. The harness resamples user
  audio to it. The default is 16 kHz.
- **`finalize()`** must produce a final transcript quickly, with
  `finalized=True`, for everything sent so far. This is what makes end of turn
  fast. Cuelo's controllers decide the turn ended, and the STT must not wait
  for its own endpointing. If the vendor has no flush message, send silence or
  close-and-reopen, and say so in the docstring.
- **Interim results** (`is_final=False`) replace each other. Finals accumulate.
- **Errors** raise `ProviderError` with the vendor's message, so they show up in
  the `error` event and the dashboard.
- **Timestamps** (`start`, `end`) are seconds from session start when the
  vendor provides them.
</Step>

<Step>
### Add an extra [#add-an-extra]

If the provider needs packages, add an optional-dependency group named after
the vendor. Reuse what's already pinned. Most providers need only `httpx` and
`websockets`.

```toml title="engine/pyproject.toml"
[project.optional-dependencies]
acme = [
    "websockets>=17.4",
]
```

Then `cd engine && uv lock`.
</Step>

<Step>
### Register it in the schema [#register-it-in-the-schema]

`schema/config.cuelo.schema.json` is the single source of truth for the CLI
and the engine. Add the name to the provider enum:

```json title="schema/config.cuelo.schema.json"
"stt": {
  "properties": {
    "provider": {
      "enum": ["smallest", "elevenlabs", "deepgram", "cartesia", "inworld", "assemblyai", "soniox", "acme"]
    }
  }
}
```

and an entry in the `x-cuelo-providers` registry, which maps the name to the
class, the extra and the credential:

```json title="schema/config.cuelo.schema.json"
"x-cuelo-providers": {
  "stt": {
    "acme": {
      "class": "cuelo.stt.acme.AcmeSTT",
      "extra": "acme",
      "env": "ACME_API_KEY"
    }
  }
}
```

- `class` is what `cuelo generate` imports and instantiates in
  `generated/providers.py`, and what the default agent uses for `CUELO_STT=acme`.
- `extra` is added to generated projects' dependencies.
- `env` is added to the project's required environment and `.env.example`,
  and checked by `cuelo doctor`. It must equal the class's `default_api_key_env`.

For an **LLM**, the enum appears three times: `$defs.llm`, `$defs.llmModel`
and `$defs.thinker`. Update all three.
</Step>

<Step>
### Sync the copies [#sync-the-copies]

```bash
make schema
```

This copies the schema into `engine/src/cuelo/config/` and
`cli/internal/config/`. Tests in both fail if the copies drift, if a registry
entry has no enum value (or the other way round), or if `env` doesn't match the
class.
</Step>

<Step>
### Add it to `cuelo init` [#add-it-to-cuelo-init]

```go title="cli/internal/cli/init.go"
func orderedNames(stage string) []string {
	switch stage {
	case "stt":
		return []string{"smallest", "elevenlabs", "deepgram", "cartesia", "inworld", "assemblyai", "soniox", "acme"}
```

For an LLM with an obvious default model, add it to `defaultModels` in the
same file, so `cuelo init --llm acme` works without `--model`.
</Step>

<Step>
### Test against a fake server [#test-against-a-fake-server]

CI never calls real APIs. Write a local fake that follows the vendor's
documented protocol (auth, URL, message shapes, finalize, close and errors)
and run your provider against it. See `engine/tests/test_more_speech_providers.py`
for the pattern used by every provider:

```python title="engine/tests/test_acme.py"
import json
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

import pytest
from websockets.asyncio.server import ServerConnection, serve

from cuelo.provider import ProviderError
from cuelo.runtime.audio import Audio
from cuelo.stt.acme import AcmeSTT

PCM = b"\x01\x00" * 1600  # 100 ms at 16 kHz

@asynccontextmanager
async def ws_server(handler) -> AsyncIterator[str]:
    async with serve(handler, "127.0.0.1", 0) as server:
        port = next(iter(server.sockets)).getsockname()[1]
        yield f"http://127.0.0.1:{port}"

async def test_acme_stt() -> None:
    seen: dict = {}

    async def handler(ws: ServerConnection) -> None:
        seen["auth"] = ws.request.headers["Authorization"]
        seen["config"] = json.loads(await ws.recv())
        audio = await ws.recv()
        seen["audio"] = len(audio)
        await ws.send(json.dumps({"type": "partial", "text": "hel"}))
        assert json.loads(await ws.recv()) == {"type": "flush"}
        await ws.send(json.dumps({"type": "final", "text": "hello", "flushed": True}))
        await ws.close()

    async with ws_server(handler) as url:
        stt = AcmeSTT(base_url=url, api_key="k", language="en-US", vocab=["Cuelo"])
        session = await stt.connect()
        await session.send(Audio(PCM, 16_000))
        await session.finalize()
        out = [t async for t in session.transcripts()]

    assert seen["auth"] == "Bearer k"
    assert seen["config"]["lang"] == "en" and seen["config"]["boost"] == ["Cuelo"]
    assert seen["audio"] == len(PCM)
    assert [(t.text, t.is_final, t.finalized) for t in out] == [
        ("hel", False, False),
        ("hello", True, True),
    ]

async def test_acme_stt_error() -> None:
    async def handler(ws: ServerConnection) -> None:
        await ws.recv()
        await ws.send(json.dumps({"type": "error", "message": "invalid model"}))

    async with ws_server(handler) as url:
        session = await AcmeSTT(base_url=url, api_key="k").connect()
        with pytest.raises(ProviderError, match="invalid model"):
            async for _ in session.transcripts():
                pass
```

Cover at least: the request (auth, URL, config message), audio framing,
interim versus final versus finalized, and an error message surfacing as
`ProviderError`. Then run `make check`.
</Step>

<Step>
### Try it live [#try-it-live]

With a real key in `.env`:

```bash
CUELO_STT=acme cuelo start
```

Talk on the Talk page and check the call in the dashboard: transcripts arrive,
`finalize` is fast (see the latency tab), and nothing errors. In your PR, say
which models and languages you tested live.
</Step>

<Step>
### Document it [#document-it]

Add a row to the provider table in
`docs/content/reference/providers.mdx`, with the name, class, extra and
variable, plus the variable in `reference/environment.mdx`. Note anything
unusual, such as unsupported settings, model quirks or rate limits.
</Step>
</Steps>

## Text-to-speech [#text-to-speech]

Choose the base class by how the vendor's API works:

| Base           | The vendor…                                      | You implement                                              |
| -------------- | ------------------------------------------------ | ---------------------------------------------------------- |
| `TTS`          | takes full text per request (HTTP, SSE, chunked) | `synthesize(text) -> AsyncIterator[Audio]`                 |
| `StreamingTTS` | takes text incrementally over one connection     | `stream(text: AsyncIterator[str]) -> AsyncIterator[Audio]` |

With `TTS`, Cuelo splits the LLM's stream into sentences and calls
`synthesize` for each, so the first sentence plays while the rest is written.
Register both under separate names when a vendor offers both, such as
`acme` and `acme-streaming`.

```python title="engine/src/cuelo/tts/acme.py"
class AcmeTTS(TTS):
    default_base_url = "https://api.acme.example"
    default_api_key_env = "ACME_API_KEY"
    supports = frozenset({"voice", "speed", "language", "vocab"})

    def __init__(self, *, voice: str, model: str = "acme-voice-1", sample_rate: int = 24_000, **kwargs):
        super().__init__(voice=voice, **kwargs)   # speed, language, vocab, base_url, api_key, api_key_env
        self.model = model
        self.sample_rate = sample_rate            # the rate of the audio you yield
        self._http: httpx.AsyncClient | None = None

    async def start(self) -> None:                # once, before the first call: fail fast, warm up
        _ = self.api_key

    async def close(self) -> None:
        if self._http is not None:
            await self._http.aclose()

    async def synthesize(self, text: str) -> AsyncIterator[Audio]:
        body = {"text": self.apply_vocab(text), "voice": self.voice, "format": "pcm_s16le",
                "sample_rate": self.sample_rate}
        if self.speed is not None:
            body["speed"] = self.speed
        async with self.http.stream("POST", "/v1/speak", json=body,
                                    headers={"Authorization": f"Bearer {self.api_key}"}) as r:
            if r.is_error:
                await r.aread()
                raise ProviderError(f"Acme TTS {r.status_code}: {r.text}")
            pcm = PCMBuffer()
            async for chunk in r.aiter_bytes():
                if data := pcm.push(chunk):       # keeps sample alignment across chunks
                    yield Audio(data, self.sample_rate)
            if data := pcm.flush():
                yield Audio(data, self.sample_rate)
```

Rules for TTS:

- Yield **PCM16 mono** at `self.sample_rate`. Strip WAV headers and decode
  base64 or μ-law in the provider. The harness resamples.
- Call `self.apply_vocab(text)` so pronunciations from the voice profile
  apply. If the vendor has native pronunciation dictionaries, use those
  instead and keep `vocab` in `supports`.
- **Cancellation:** on barge-in the harness closes your iterator. Use
  `async with` and `try/finally` so connections and requests close. A
  streaming provider must stop the vendor's generation (send its cancel or
  close message) so it isn't billed for audio no one hears.
- For streaming providers over WebSocket, `cuelo.tts.sockets.WarmSocket`
  keeps a connection open between turns, which saves a handshake on every reply.
- Add `cli/internal/voices/<vendor>.go` (step 7) implementing `List` and
  `Preview`, and a case in `ForTTS`, so `cuelo voices acme` works.

## LLMs [#llms]

If the vendor speaks the OpenAI Chat Completions API, a provider is three lines:

```python title="engine/src/cuelo/llms/acme.py"
"""Acme: fast inference behind an OpenAI-compatible endpoint.

Docs: https://docs.acme.example/api
"""

from cuelo.llms.openai_spec import OpenAISpecLLM

class AcmeLLM(OpenAISpecLLM):
    default_base_url = "https://api.acme.example/v1"
    default_api_key_env = "ACME_API_KEY"
```

Register it with `"extra": "openai"`. Otherwise implement `stream()`:

```python
class AcmeLLM(LLM):
    def stream(self, messages, tools=()) -> AsyncIterator[TextDelta | ToolCall | Done]:
        ...
```

- Convert `Message`s (roles `system`, `user`, `assistant`, `tool`; parts
  `TextPart`, `ToolCall`, `ToolResult`) and `Tool`s (JSON Schema parameters)
  to the vendor's format. `anthropic.py` is a complete native example.
- Yield `TextDelta(text)` as tokens arrive. Yield a `ToolCall(id, name, arguments)`
  once its arguments are complete. Finish with exactly one
  `Done(stop_reason, usage=Usage(input_tokens, output_tokens))`.
- Closing the iterator must cancel the request, because barge-in does exactly that.
- Time to first token is the number that matters. Don't buffer.

## Turn-taking models [#turn-taking-models]

Controllers are providers too, without credentials. They have their own guide:
[Add a turn-taking model](/contributing/controllers).

## Checklist [#checklist]

- [ ] Module docstring: protocol summary, how `finalize`/cancel works, setting mappings, doc links
- [ ] `default_base_url`, `default_api_key_env`, `supports` set
- [ ] Errors raise `ProviderError` with the vendor's message
- [ ] Extra in `pyproject.toml`, `uv lock`
- [ ] Enum and registry entry in the schema (all three LLM enums), `make schema`
- [ ] `orderedNames` (and `defaultModels` for LLMs) in `init.go`
- [ ] Fake-server tests. `make check` passes.
- [ ] Tried live. Models and languages listed in the PR.
- [ ] Docs: providers table and environment variables