Skip to content

API reference

This page is generated automatically from the docstrings in the source.

Package

wavecat_sdk

Wavecat SDK — connect your own OpenAI-compatible backend to wavecat.

The SDK is a small, always-on gateway: it presents a stable OpenAI-compatible endpoint (/v1/chat/completions, /v1/models, /health) on localhost and forwards every request to your model server (llama.cpp, vLLM, …). Point wavecat's "Custom backend" Base URL at this gateway and it routes its heavy-model work here instead of the local 35B.

The gateway only ever sees OpenAI chat-completion payloads. It never runs wavecat tools and never touches user data — tools always execute inside wavecat; this process only generates tokens.

Configuration

wavecat_sdk.config

Gateway configuration: where to listen and which upstream to forward to.

Settings dataclass

Runtime config for the gateway.

Attributes:

Name Type Description
upstream_url str

The OpenAI API root of YOUR model server (llama.cpp / vLLM), e.g. http://127.0.0.1:8000/v1.

upstream_key str

Optional bearer token your server expects.

model str | None

If set, the inbound model field is rewritten to this before forwarding — so wavecat can send any model id and you decide what actually serves it. None passes the inbound model through.

host / port

Where the gateway itself listens (wavecat points here).

strip_keys tuple[str, ...]

Top-level request-body keys to drop before forwarding. These are wavecat/llama.cpp-isms a stricter server (vLLM) may reject.

request_timeout float

Per-request upstream timeout (seconds).

Source code in src/wavecat_sdk/config.py
@dataclass
class Settings:
    """Runtime config for the gateway.

    Attributes:
        upstream_url: The OpenAI API root of YOUR model server (llama.cpp / vLLM),
            e.g. ``http://127.0.0.1:8000/v1``.
        upstream_key: Optional bearer token your server expects.
        model: If set, the inbound ``model`` field is rewritten to this before
            forwarding — so wavecat can send any model id and you decide what
            actually serves it. ``None`` passes the inbound model through.
        host / port: Where the gateway itself listens (wavecat points here).
        strip_keys: Top-level request-body keys to drop before forwarding. These
            are wavecat/llama.cpp-isms a stricter server (vLLM) may reject.
        request_timeout: Per-request upstream timeout (seconds).
    """

    upstream_url: str = "http://127.0.0.1:8000/v1"
    upstream_key: str = ""
    model: str | None = None
    host: str = "127.0.0.1"
    port: int = 8800
    strip_keys: tuple[str, ...] = DEFAULT_STRIP_KEYS
    request_timeout: float = 600.0

    def __post_init__(self) -> None:
        self.upstream_url = self.upstream_url.rstrip("/")

    def upstream(self, path: str) -> str:
        """Build a full upstream URL for an OpenAI sub-path (e.g. ``/chat/completions``)."""
        return f"{self.upstream_url}/{path.lstrip('/')}"

    @property
    def auth_headers(self) -> dict[str, str]:
        return {"Authorization": f"Bearer {self.upstream_key}"} if self.upstream_key else {}

upstream

upstream(path: str) -> str

Build a full upstream URL for an OpenAI sub-path (e.g. /chat/completions).

Source code in src/wavecat_sdk/config.py
def upstream(self, path: str) -> str:
    """Build a full upstream URL for an OpenAI sub-path (e.g. ``/chat/completions``)."""
    return f"{self.upstream_url}/{path.lstrip('/')}"

from_env

from_env() -> Settings

Build :class:Settings from WAVECAT_SDK_* env vars (CLI flags override).

Source code in src/wavecat_sdk/config.py
def from_env() -> Settings:
    """Build :class:`Settings` from ``WAVECAT_SDK_*`` env vars (CLI flags override)."""
    strip = os.getenv("WAVECAT_SDK_STRIP_KEYS")
    strip_keys = tuple(k.strip() for k in strip.split(",") if k.strip()) if strip else DEFAULT_STRIP_KEYS
    return Settings(
        upstream_url=os.getenv("WAVECAT_SDK_UPSTREAM", "http://127.0.0.1:8000/v1"),
        upstream_key=os.getenv("WAVECAT_SDK_KEY", ""),
        model=os.getenv("WAVECAT_SDK_MODEL") or None,
        host=os.getenv("WAVECAT_SDK_HOST", "127.0.0.1"),
        port=int(os.getenv("WAVECAT_SDK_PORT", "8800")),
        strip_keys=strip_keys,
        request_timeout=float(os.getenv("WAVECAT_SDK_TIMEOUT", "600")),
    )

Gateway

wavecat_sdk.gateway

The OpenAI-compatible proxy gateway wavecat talks to.

wavecat ──/v1──► [this gateway] ──/v1──► your llama.cpp / vLLM

Three routes are exposed:

  • POST /v1/chat/completions — proxy (streaming + non-streaming) to your upstream. The inbound body is lightly sanitized (drop wavecat/llama.cpp-only keys a strict server might reject; optionally rewrite the model id).
  • GET /v1/models — proxied so wavecat's reachability probe + Test button verify the WHOLE chain (gateway → upstream), not just the gateway.
  • GET /health — gateway liveness.

The gateway never executes tools and never sees wavecat internals — it only relays OpenAI chat payloads (prompts, tool schemas/results-as-text, sampling params) to and from your model.

create_app

create_app(settings: Settings) -> FastAPI

Build the gateway app bound to a given upstream.

Source code in src/wavecat_sdk/gateway.py
def create_app(settings: Settings) -> FastAPI:
    """Build the gateway app bound to a given upstream."""
    client = httpx.AsyncClient(timeout=settings.request_timeout)

    @asynccontextmanager
    async def lifespan(_: FastAPI) -> AsyncIterator[None]:
        # Own the upstream HTTP client for the app's lifetime; close it on shutdown.
        try:
            yield
        finally:
            await client.aclose()

    app = FastAPI(title="wavecat-sdk gateway", lifespan=lifespan)

    @app.get("/health")
    async def health() -> dict[str, str]:
        return {"status": "ok", "upstream": settings.upstream_url}

    @app.get("/v1/models")
    async def models() -> Response:
        """Proxy the upstream model list (so a probe checks the full chain)."""
        try:
            r = await client.get(settings.upstream("/models"), headers=settings.auth_headers)
        except httpx.RequestError as exc:
            logger.warning("upstream unreachable on /v1/models: %s", exc)
            return JSONResponse({"error": f"upstream unreachable: {exc}"}, status_code=503)
        return Response(content=r.content, status_code=r.status_code, media_type="application/json")

    @app.post("/v1/chat/completions")
    async def chat_completions(request: Request) -> Response:
        """Proxy a chat completion, streaming when the client asked for it."""
        raw = await request.body()
        try:
            body = json.loads(raw or b"{}")
        except json.JSONDecodeError:
            return JSONResponse({"error": "invalid JSON body"}, status_code=400)

        body = _sanitize(body, settings)
        stream = bool(body.get("stream"))
        url = settings.upstream("/chat/completions")
        headers = {"Content-Type": "application/json", **settings.auth_headers}

        if not stream:
            try:
                r = await client.post(url, json=body, headers=headers)
            except httpx.RequestError as exc:
                logger.warning("upstream unreachable on /v1/chat/completions: %s", exc)
                return JSONResponse({"error": f"upstream unreachable: {exc}"}, status_code=503)
            return Response(
                content=r.content,
                status_code=r.status_code,
                media_type=r.headers.get("content-type", "application/json"),
            )

        async def relay() -> Any:
            # Stream the upstream SSE through unchanged so wavecat's reader sees
            # the exact OpenAI delta frames it expects.
            try:
                async with client.stream("POST", url, json=body, headers=headers) as up:
                    async for chunk in up.aiter_raw():
                        if chunk:
                            yield chunk
            except httpx.RequestError as exc:
                logger.warning("upstream unreachable mid-stream: %s", exc)
                err = {"error": {"message": f"upstream unreachable: {exc}", "type": "gateway"}}
                yield f"data: {json.dumps(err)}\n\n".encode()
                yield b"data: [DONE]\n\n"

        return StreamingResponse(relay(), media_type="text/event-stream")

    return app

CLI

wavecat_sdk.cli

wavecat-sdk command line — run the gateway.

Example::

wavecat-sdk serve --upstream http://127.0.0.1:8000/v1 --model my-model --port 8800

Then in wavecat → Settings → Backend, enable the custom backend and set the Base URL to http://127.0.0.1:8800/v1.