# Python API (/reference/python) Generated projects use these classes, and you can use them directly to build an agent without `config.cuelo`. The **slot interfaces** (`STT`, `LLM`, `TTS`, `Controller`, `TurnTakingPolicy`) are public API. ## A whole agent in code [#a-whole-agent-in-code] ```python title="agent.py" from cuelo.controllers.turn1_mini import Turn1MiniController from cuelo.controllers.backchannel_detector import BackchannelDetectorController from cuelo.database.sql import SQLStore from cuelo.llms.cerebras import CerebrasLLM from cuelo.runtime.harness import Harness, Pipeline from cuelo.stt.smallest import SmallestSTT from cuelo.tts.smallest import SmallestTTS def build() -> Pipeline: return Pipeline( stt=SmallestSTT(language="en"), llm=CerebrasLLM("qwen-3.8-27b", max_tokens=512), tts=SmallestTTS(voice="aarini"), controllers=[Turn1MiniController(), BackchannelDetectorController()], instructions="You are a concise barista. One or two sentences per reply.", greeting="Hi! What can I get you?", ) harness = Harness(build, agent="barista", store=SQLStore("sqlite://cuelo.db")) ``` ```bash cuelo-serve agent.py # Realtime API + dashboard on :4141 ``` `cuelo-serve` loads any file that defines `harness`. Optional module-level names: `routes` (a function `harness -> list[BaseRoute]`), `realtime` (bool), `api_keys` and `cors` (lists). ## `Harness` [#harness] ```python Harness( build: Callable[[], Pipeline], # a fresh Pipeline per call *, agent: str = "agent", # name shown in the dashboard store: Store | None = None, # where events are persisted assets: str | Path = "assets", # cache for pre-synthesized phrases recordings: str | Path | None = "recordings", input_sample_rate: int = 24_000, output_sample_rate: int = 24_000, on_event: EventHook | None = None, # every event of every call, sync or async ) ``` | Member | | | ------------------------------------------------- | ----------------------------------- | | `await harness.open(session_id=None)` | start a call and return a `Session` | | `harness.sessions` | live sessions by id | | `harness.agent` | agent name | | `harness.input_sample_rate`, `output_sample_rate` | PCM rates at the edge | ## `Session` [#session] | Method | | | ------------------------------------------------ | ---------------------------------------------------------------------------------------------- | | `await send_audio(audio: Audio)` | user audio in | | `output() -> AsyncIterator[AgentAudio \| Event]` | agent audio and events, in order | | `await respond(text=None, *, speak_first=False)` | reply now: to `text` (typed), to what was said so far (push-to-talk), or first (`speak_first`) | | `await say(text)` | speak `text` now as an agent turn, no LLM. Returns immediately. | | `await interrupt()` | stop the agent | | `await set_voice(voice)` | switch TTS voice, re-rendering backchannel phrases | | `await close(reason="closed")` | end the call | | `id`, `pipeline` | | ## `Pipeline` [#pipeline] The components of one call. A dataclass. Change fields in `customize()`. | Field | Type | | | -------------- | ------------------------ | ----------------------------------- | | `stt` | `STT` | | | `llm` | `LLM` | the talker | | `tts` | `BaseTTS` | | | `controllers` | `list[Controller]` | turn-taking | | `policy` | `TurnTakingPolicy` | default `StandardTurnTaking()` | | `instructions` | `str` | system prompt | | `tools` | `list[FunctionTool]` | server-side tools | | `client_tools` | `list[Tool]` | tools the Realtime client runs | | `thinker` | `LLM \| Thinker \| None` | enables thinker-talker | | `summarizer` | `LLM \| None` | writes `summary.created` at the end | | `greeting` | `str \| None` | said as soon as the call starts | ## `Audio` [#audio] ```python Audio(pcm: bytes, sample_rate: int, channels: int = 1) # int16 little-endian audio.samples, audio.duration ``` ## Slot interfaces [#slot-interfaces] ### `STT`: `cuelo.stt.interface` [#stt-cuelosttinterface] ```python class STT(Provider): sample_rate: int = 16_000 # the harness resamples to this supports: ClassVar[frozenset["language" | "vocab"]] async def connect(self) -> STTSession: ... class STTSession: async def send(self, audio: Audio) -> None: ... async def finalize(self) -> None: ... # force a final for everything sent def transcripts(self) -> AsyncIterator[Transcript]: ... async def close(self) -> None: ... Transcript(text, is_final, finalized=False, start=None, end=None, language=None) ``` ### `LLM`: `cuelo.llms.interface` [#llm-cuelollmsinterface] ```python class LLM(Provider): accepts_audio: bool = False def __init__(self, model, *, temperature=None, top_p=None, max_tokens=None, base_url=None, api_key=None, api_key_env=None): ... def stream(self, messages: Sequence[Message], tools: Sequence[Tool] = ()) -> AsyncIterator[TextDelta | ToolCall | Done]: ... async def complete(self, messages, tools=()) -> tuple[str, list[ToolCall], Done]: ... ``` `stream` yields `TextDelta`s and `ToolCall`s, then exactly one `Done`. Closing the iterator cancels generation. ### `TTS`: `cuelo.tts.interface` [#tts-cuelottsinterface] ```python class BaseTTS(Provider): sample_rate: int supports: ClassVar[frozenset["voice" | "speed" | "language" | "vocab"]] voice, speed, language, vocab # changeable between utterances def stream(self, text: AsyncIterator[str]) -> AsyncIterator[Audio]: ... class TTS(BaseTTS): # full text in, audio out def synthesize(self, text: str) -> AsyncIterator[Audio]: ... class StreamingTTS(BaseTTS): # text in and audio out at the same time def stream(self, text: AsyncIterator[str]) -> AsyncIterator[Audio]: ... ``` The harness always calls `stream()`. `TTS` gets one for free that synthesizes sentence by sentence. On barge-in the iterator is closed, so clean up in `finally`. ### `Controller`: `cuelo.controllers.interface` [#controller-cuelocontrollersinterface] ```python class Controller(ABC): duplex: ClassVar[bool] = False # also hears the agent (push_agent) emits: ClassVar[frozenset[Signal]] sample_rate: int = 16_000 agent_speaking: bool # set by the harness def push_user(self, audio: Audio) -> list[ControlEvent]: ... # abstract def push_agent(self, audio: Audio) -> list[ControlEvent]: ... def push_transcript(self, t: Transcript) -> list[ControlEvent]: ... def reset(self) -> None: ... Signal: TURN_END | INTERRUPT | USER_BACKCHANNEL | BACKCHANNEL ControlEvent(signal, t, text=None, confidence=1.0, speech_end_t=None) ``` Controllers are synchronous. The harness runs each on its own worker thread. `t` is seconds of user audio consumed, so runs are replayable. ### `TurnTakingPolicy`: `cuelo.policies.turn_taking` [#turntakingpolicy-cuelopoliciesturn_taking] ```python class TurnTakingPolicy(ABC): def on_event(self, event: ControlEvent) -> list[Decision]: ... def tick(self, t: float) -> list[Decision]: ... # timers def agent_started(self, t: float) -> None: ... def agent_stopped(self, t: float) -> None: ... Action: RESPOND | STOP | BACKCHANNEL | CONTINUE ``` ### `Provider`: `cuelo.provider` [#provider-cueloprovider] The base of every STT, LLM and TTS: ```python class Provider: default_base_url: ClassVar[str | None] default_api_key_env: ClassVar[str | None] def __init__(self, *, base_url=None, api_key=None, api_key_env=None): ... api_key: str # resolved, raises MissingCredentialError async def start(self) -> None: ... # warm up once async def close(self) -> None: ... ``` ## Tools [#tools] ```python from cuelo.runtime.tools import FunctionTool, function_tool @function_tool(description="...", parameters={...}) # description defaults to the docstring async def my_tool(**kwargs) -> dict | str: ... ``` ## Serving [#serving] ```python from cuelo.serve import create_app app = create_app(harness, routes=[...], realtime=True, api_keys=[...], cors=[...]) ``` `create_app` returns a Starlette app. Run it with any ASGI server.