client.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/example/client.py
 
"""
ExampleClient — deterministic stub that satisfies :class:`BaseClient`.
 
Purpose
-------
SDK-free and network-free. ``stream_chat`` and ``one_shot`` build a
fake response from the last user message ("echo: <text>"). Useful as
a template for new provider clients and as a test harness for code
that needs a working :class:`BaseClient` without hitting a real API.
``oc_diagnostics`` is the only external import — it's first-party
Option C tooling, not a foreign SDK.
 
When porting this file to a real provider:
 
  1. Replace the constructor body with the SDK client construction
     (e.g. ``self._sdk = OpenAI(api_key=..., base_url=...)``).
  2. Replace ``stream_chat`` with a real streaming loop that yields
     ``("reasoning", str)``, ``("content", str)``,
     ``("tool_call", dict)``, and ``("usage", dict)`` chunks as the
     SDK produces them. See ``providers/deepseek/client.py`` for the
     canonical shape.
  3. Replace ``one_shot`` with a non-streaming completion call.
  4. Replace ``list_models`` with a real API call returning model ids.
 
Responsibilities
----------------
- :class:`ExampleClient` — :class:`BaseClient` stub with no I/O.
- :func:`_last_user_text` — pluck the most recent user message
  content from the messages list.
- Deterministic fake usage block (constant token counts) so tests
  can assert on it.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- ``stream_chat`` always yields exactly one ``("content", str)``
  chunk and one ``("usage", dict)`` chunk — no reasoning, no tool
  calls.
- ``one_shot`` returns a plain ``str`` — never ``None``.
- ``list_models`` returns the static :data:`MODELS` keys.
- No exceptions are raised under normal use; the stub is meant to
  be reliable in tests.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- providers.example.models (MODELS registry whose keys list_models returns)
- option_c_agent_contracts (lazy; SupportsCancellation type of the stream_chat cancellation token)
"""
from __future__ import annotations
 
from typing import Any, Iterator, TYPE_CHECKING
 
from oc_diagnostics import _dbg
 
if TYPE_CHECKING:
    from option_c_agent_contracts import SupportsCancellation
 
from .models import MODELS
 
 
def _last_user_text(messages: list[dict]) -> str:
    """Return the most recent user message's content, or empty string."""
    for msg in reversed(messages):
        if msg.get("role") == "user":
            content = msg.get("content")
            if isinstance(content, str):
                return content
    return ""
 
 
class ExampleClient:
    """Stub :class:`BaseClient` for the example provider.
 
    Stores the constructor arguments so tests can assert on them, but
    does nothing with them otherwise.
    """
 
    # OCT-LINT: disable=no_hardcoded_secrets reason="Empty-string default for optional api_key param, not a credential"
    def __init__(self, api_key: str = "", base_url: str = "") -> None:
        func = "__init__"
        self.api_key = api_key
        self.base_url = base_url
        _dbg("OC_CHAT.CLIENT", func, f"ExampleClient base_url={base_url!r}", 4)
 
    # ── Streaming chat ──────────────────────────────────────────────────────
 
    def stream_chat(
        self,
        messages: list[dict],
        model: str,
        *,
        effort: str | None = None,
        extra_body: dict | None = None,
        tools: list[dict] | None = None,
        max_tokens: int | None = None,
        cancellation: "SupportsCancellation | None" = None,
    ) -> Iterator[tuple[str, Any]]:
        """Yield a fixed sequence: one ``content`` chunk, then ``usage``.
 
        Real providers stream multiple ``content`` chunks plus optional
        ``reasoning`` and ``tool_call`` chunks before the final
        ``usage`` chunk. See ``providers/deepseek/client.py``.
 
        If *cancellation* is provided, the token is polled before each
        yield; a cancelled token raises :class:`CancellationError` and
        no further chunks are produced (Phase O3c).
        """
        text = f"echo: {_last_user_text(messages)}"
        if cancellation is not None:
            cancellation.raise_if_cancelled()
        yield ("content", text)
        if cancellation is not None:
            cancellation.raise_if_cancelled()
        yield ("usage", {
            "prompt_tokens": len(_last_user_text(messages)),
            "completion_tokens": len(text),
            "total_tokens": len(_last_user_text(messages)) + len(text),
        })
 
    # ── One-shot (non-streaming) ────────────────────────────────────────────
 
    def one_shot(
        self,
        messages: list[dict],
        model: str,
        *,
        effort: str | None = None,
        extra_body: dict | None = None,
        max_tokens: int | None = None,
    ) -> str:
        """Return ``"echo: <last user msg>"`` synchronously."""
        return f"echo: {_last_user_text(messages)}"
 
    # ── Model listing ───────────────────────────────────────────────────────
 
    def list_models(self) -> list[str]:
        """Return the curated example model ids (no network call)."""
        return list(MODELS.keys())