provider.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# providers/example/provider.py
 
"""
ExampleProvider — minimal :class:`ProviderPlugin` implementation.
 
Purpose
-------
Every abstract method is implemented with the simplest body that
satisfies the contract: no provider-specific quirks, no
``extra_body`` shaping, no message transformation. When porting to a
real provider, replace each method body in turn — the comments mark
every DeepSeek-specific pattern that does *not* belong in a generic
provider.
 
Responsibilities
----------------
- :class:`ExampleProvider` — implement every abstract method on
  :class:`ProviderPlugin` with the minimal body needed.
- :meth:`build_client` — return an :class:`ExampleClient`.
- :meth:`models` / :meth:`tier` — surface the static registries.
- :meth:`build_extra_body` — always return ``None`` (no thinking
  payload).
- :meth:`transform_outgoing_messages` — pass-through (no schema
  embedding).
- :meth:`parse_usage` — defer to the shared
  :func:`usage_prompt_cache_hit_miss` helper.
 
Diagnostics
-----------
Domain: OC_CHAT.CLIENT
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- The example bundle stays dependency-free — no SDK imports, no
  network code.
- ``build_extra_body`` returns ``None`` regardless of model
  capabilities (no thinking shape needed for stubs).
- ``transform_outgoing_messages`` is a true pass-through (no
  sentinel embedding) — the example provider exists to prove the
  contract works without provider-specific transforms.
- All abstract surface lives here; everything else in the bundle is
  pure data.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- option_c_agent_contracts (ProviderPlugin base class and the ModelCaps, UsageStats and BaseClient types)
- providers.example.client (ExampleClient returned by build_client)
- providers.example.models (MODELS and TIER_MAP registries)
"""
from __future__ import annotations
 
from oc_diagnostics import _dbg
 
from option_c_agent_contracts import (
    ProviderPlugin,
    ModelCaps,
    UsageStats,
    BaseClient,
)
 
from .client import ExampleClient
from .models import MODELS, TIER_MAP
 
 
class ExampleProvider(ProviderPlugin):
    name = "example"
    # OCT-LINT: disable=no_hardcoded_secrets reason="Env var NAME to read the key from, not a credential"
    api_key_env_var = "EXAMPLE_API_KEY"
    base_url_env_var = "EXAMPLE_BASE_URL"
    default_base_url = ""
    default_model = "example-echo"
 
    # ── Client construction ──────────────────────────────────────────────────
 
    def build_client(self, api_key: str, base_url: str) -> BaseClient:
        func = "build_client"
        _dbg("OC_CHAT.CLIENT", func, f"ExampleProvider base_url={base_url!r}", 4)
        return ExampleClient(api_key=api_key, base_url=base_url)
 
    # ── Model registry ───────────────────────────────────────────────────────
 
    def models(self) -> dict[str, ModelCaps]:
        return MODELS
 
    def fetch_live_models(self, client: BaseClient) -> list[str]:
        # Real providers query the live API: ``return client.list_models()``.
        # The example client returns the curated list deterministically.
        return client.list_models()
 
    def default_caps(self) -> ModelCaps:
        # Permissive default for any model that shows up in fetch_live_models
        # but is missing from MODELS. Tighten per-provider as appropriate.
        return ModelCaps(
            supports_thinking=False,
            supports_reasoning=False,
            supports_tools=True,
            max_context_tokens=4096,
        )
 
    # ── Tier mapping ─────────────────────────────────────────────────────────
 
    def tier(self, name: str) -> str | None:
        return TIER_MAP.get(name)
 
    # ── Per-call shaping ─────────────────────────────────────────────────────
 
    def build_extra_body(
        self,
        model: str,
        *,
        enable_thinking: bool,
        effort: str | None,
    ) -> dict | None:
        """No provider-specific ``extra_body`` payload.
 
        DeepSeek returns ``{"thinking": {"type": "enabled"}}`` here when
        thinking is requested and the model supports it; other providers
        with no equivalent concept return ``None``.
        """
        return None
 
    def transform_outgoing_messages(
        self,
        messages: list[dict],
        tool_schemas: list[dict] | None,
    ) -> list[dict]:
        """No-op — return messages unchanged.
 
        DeepSeek embeds JSON tool schemas into the system prompt for
        prefix-cache stability and appends a sentinel comment. Providers
        without prefix caching (or with server-side automatic caching)
        return the messages unchanged.
        """
        return messages
 
    # ── Response parsing ─────────────────────────────────────────────────────
 
    def parse_usage(self, usage: dict) -> UsageStats:
        if not isinstance(usage, dict):
            return UsageStats()
        return UsageStats(
            prompt_tokens=int(usage.get("prompt_tokens") or 0),
            completion_tokens=int(usage.get("completion_tokens") or 0),
            total_tokens=int(usage.get("total_tokens") or 0),
            # No prefix-cache accounting — leave cache fields zero.
        )