add-provider

add-provider

A provider added wrong leaks its API shape into your business logic. One API change from the vendor → rewrite half your service. Follow this checklist.

Step 1 — Decide: file or folder?

Principle A4: one file per format when a provider handles multiple formats.

Step 2 — Create the provider file from this template

# app/providers/<name>.py  (or app/providers/<name>/image.py)
from __future__ import annotations

import logging
from decimal import Decimal

import httpx

from app.core.bulkhead import get_provider_client   # Principle B5
from app.core.context import provider_ctx           # Principle B7
from app.providers.base import AIProvider, JobRequest, JobResult
from app.providers.exceptions import (
    ProviderError,
    ProviderInvalidResponseError,
    ProviderRateLimitError,
    ProviderTimeoutError,
)

logger = logging.getLogger(__name__)


class <Name>Provider:
    provider_name = "<name>"                        # used in logs + bulkhead key

    async def generate(self, request: JobRequest) -> JobResult:
        provider_ctx.set(self.provider_name)        # Principle B7: set before any I/O
        client = get_provider_client(self.provider_name)  # Principle B5: isolated client

        try:
            response = await client.post(
                "/v1/generate",
                json=self._build_payload(request),
                headers={"X-Idempotency-Key": request.idempotency_key},  # Principle B6
                timeout=30.0,
            )
            response.raise_for_status()
        except httpx.TimeoutException as e:
            raise ProviderTimeoutError(
                message=str(e), provider=self.provider_name, retryable=True
            ) from e
        except httpx.HTTPStatusError as e:
            self._map_http_error(e)

        return self._parse_response(response.json(), request)  # Principle B3: ACL here

    def _build_payload(self, request: JobRequest) -> dict:
        return {"prompt": request.prompt, "model": request.model_id, **request.params}

    def _parse_response(self, data: dict, request: JobRequest) -> JobResult:
        # ACL: validate and map to our domain type. Never return raw data.
        try:
            url = data["output"]["url"]          # adjust to actual provider shape
            cost = Decimal(str(data.get("cost", "0")))
        except (KeyError, TypeError) as e:
            raise ProviderInvalidResponseError(
                message=f"Unexpected response shape: {e}",
                provider=self.provider_name,
                retryable=False,
                raw_response=data,
            ) from e
        return JobResult(
            url=url,
            cost_usd=cost,
            provider=self.provider_name,
            model_id=request.model_id,
        )

    def _map_http_error(self, e: httpx.HTTPStatusError) -> None:
        if e.response.status_code == 429:
            retry_after = int(e.response.headers.get("Retry-After", 60))
            raise ProviderRateLimitError(
                message="Rate limited",
                provider=self.provider_name,
                retryable=True,
                retry_after=retry_after,
            ) from e
        raise ProviderError(
            message=f"HTTP {e.response.status_code}",
            provider=self.provider_name,
            retryable=e.response.status_code >= 500,
        ) from e