Python API

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"))

cuelo-serve agent.py          # Realtime API + dashboard on :4141

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
)

Audio(pcm: bytes, sample_rate: int, channels: int = 1)   # int16 little-endian
audio.samples, audio.duration

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)

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]: ...

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]: ...

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)

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

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

from cuelo.runtime.tools import FunctionTool, function_tool

@function_tool(description="...", parameters={...})   # description defaults to the docstring
async def my_tool(**kwargs) -> dict | str: ...

from cuelo.serve import create_app
app = create_app(harness, routes=[...], realtime=True, api_keys=[...], cors=[...])