python-external-client

External HTTP client pattern

Every external system your service talks to — an LLM provider, Stripe, Slack, a CRM — gets its own typed Python wrapper. The wrapper is a thin class with one method per API call your service actually needs. Inside, it uses a shared httpx.AsyncClient pool that lives for the app's lifetime and is torn down at shutdown.

This skill encodes the canonical shape: a BaseHTTPClient for shared concerns (retry policy, error handling), per-system subclasses with domain-shaped methods, and httpx.MockTransport for fast in-process tests.

Bonus payoff: provider decoupling. When you have multiple providers for the same role (OpenAI / Anthropic / Gemini for LLM, SendGrid / Postmark / SES for email), define a Protocol for the role and let each concrete client satisfy it. Services depend on the Protocol; swapping providers is one env var.

When to use this skill

• Adding a new external client (app/clients/<name>.py)
• Adding a method to an existing client because a service needs one new call
• Writing tests for a client (or a service that uses one)
• Choosing a retry policy

The rules

1. BaseHTTPClient carries the retry policy and the http pool

# app/clients/base.py — shape; full version in examples/base.py
type HTTPMethod = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"]


class BaseHTTPClient:
    def __init__(self, *, http: httpx.AsyncClient, config: BaseHTTPClientConfig) -> None:
        self._http = http
        self._config = config

    @classmethod
    def from_config(cls, config: BaseHTTPClientConfig) -> Self:
        http = httpx.AsyncClient(
            base_url=config.BASE_URL,
            timeout=config.TIMEOUT,
            limits=httpx.Limits(
                max_keepalive_connections=config.KEEP_ALIVE_CONNECTIONS,
                max_connections=config.MAX_CONNECTIONS,
            ),
            headers=config.auth_headers(),     # per-system seam, see config below
        )
        return cls(http=http, config=config)

    async def close(self) -> None:
        await self._http.aclose()

    async def _request(self, method: HTTPMethod, url: str, **kwargs: Any) -> httpx.Response:
        # tenacity AsyncRetrying with stop_after_attempt(config.MAX_RETRIES) +
        # wait_exponential, retrying on TransportError and HTTPStatusError.
        ...