service-client

name	service-client
description	Generate typed async httpx clients for inter-service communication. Use when: (1) Calling other microservices, (2) Need retry logic and circuit breakers, (3) Want typed request/response models, (4) Implementing service discovery patterns. Generates async httpx clients with Pydantic models, error handling, and observability. NOT for external API integrations (use adapter-interface instead).

Service Client Generator

Generate typed async HTTP clients for microservice communication.

Quick Reference

/service-client persistence    # Client for Persistence Service
/service-client llm-gateway    # Client for LLM Gateway
/service-client <service>      # Client for any service

Generated Structure

clients/
├── base.py                    # Base client with retry logic
├── persistence_client.py      # Typed client for service
└── models/
    └── persistence.py         # Request/response models

Implementation

1. Base Client

# clients/base.py
import httpx
from typing import TypeVar, Type
from pydantic import BaseModel
import structlog

logger = structlog.get_logger()
T = TypeVar("T", bound=BaseModel)

class ServiceClient:
    """Base client with retry and error handling."""

    def __init__(
        self,
        base_url: str,
        timeout: float = 30.0,
        retries: int = 3,
    ):
        self.base_url = base_url.rstrip("/")
        self.timeout = timeout
        self.retries = retries
        self._client: httpx.AsyncClient | None = None

    async def _get_client(self) -> httpx.AsyncClient:
        if self._client is None:
            self._client = httpx.AsyncClient(
                base_url=self.base_url,
                timeout=self.timeout,
            )
        return self._client

    async def close(self) -> None:
        if self._client:
            await self._client.aclose()
            self._client = None

    async def _request(
        self,
        method: str,
        path: str,
        response_model: Type[T] | None = None,
        **kwargs,
    ) -> T | dict:
        client = await self._get_client()
        last_error = None

        for attempt in range(self.retries):
            try:
                response = await client.request(method, path, **kwargs)
                response.raise_for_status()
                data = response.json()

                if response_model:
                    return response_model.model_validate(data)
                return data

            except httpx.HTTPStatusError as e:
                if e.response.status_code < 500:
                    raise  # Don't retry client errors
                last_error = e
                logger.warning("request_failed", attempt=attempt, error=str(e))

            except httpx.RequestError as e:
                last_error = e
                logger.warning("request_error", attempt=attempt, error=str(e))

        raise last_error

    async def get(self, path: str, response_model: Type[T] | None = None, **kwargs) -> T | dict:
        return await self._request("GET", path, response_model, **kwargs)

    async def post(self, path: str, response_model: Type[T] | None = None, **kwargs) -> T | dict:
        return await self._request("POST", path, response_model, **kwargs)

    async def put(self, path: str, response_model: Type[T] | None = None, **kwargs) -> T | dict:
        return await self._request("PUT", path, response_model, **kwargs)

    async def delete(self, path: str, **kwargs) -> None:
        await self._request("DELETE", path, **kwargs)

2. Typed Service Client

# clients/persistence_client.py
from clients.base import ServiceClient
from clients.models.persistence import (
    Conversation,
    ConversationCreate,
    Message,
    MessageCreate,
    PaginatedResponse,
)

class PersistenceClient(ServiceClient):
    """Typed client for Persistence Service."""

    # Conversations
    async def create_conversation(
        self,
        data: ConversationCreate,
    ) -> Conversation:
        return await self.post(
            "/conversations",
            response_model=Conversation,
            json=data.model_dump(),
        )

    async def get_conversation(self, id: str) -> Conversation:
        return await self.get(f"/conversations/{id}", response_model=Conversation)

    async def list_conversations(
        self,
        user_id: str,
        limit: int = 20,
        offset: int = 0,
    ) -> PaginatedResponse[Conversation]:
        return await self.get(
            "/conversations",
            response_model=PaginatedResponse[Conversation],
            params={"user_id": user_id, "limit": limit, "offset": offset},
        )

    # Messages
    async def add_message(
        self,
        conversation_id: str,
        data: MessageCreate,
    ) -> Message:
        return await self.post(
            f"/conversations/{conversation_id}/messages",
            response_model=Message,
            json=data.model_dump(),
        )

3. Models

# clients/models/persistence.py
from pydantic import BaseModel
from datetime import datetime
from typing import Generic, TypeVar

T = TypeVar("T")

class ConversationCreate(BaseModel):
    user_id: str
    title: str | None = None

class Conversation(BaseModel):
    id: str
    user_id: str
    title: str | None
    created_at: datetime
    updated_at: datetime

class MessageCreate(BaseModel):
    role: str
    content: str

class Message(BaseModel):
    id: str
    conversation_id: str
    role: str
    content: str
    created_at: datetime

class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    limit: int
    offset: int

Usage

from clients.persistence_client import PersistenceClient

client = PersistenceClient(base_url="http://persistence:8000")

# Create conversation
conv = await client.create_conversation(
    ConversationCreate(user_id="user123", title="New Chat")
)

# Add message
msg = await client.add_message(
    conv.id,
    MessageCreate(role="user", content="Hello!")
)

FastAPI Integration

from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.persistence = PersistenceClient(settings.PERSISTENCE_URL)
    yield
    await app.state.persistence.close()

def get_persistence(request: Request) -> PersistenceClient:
    return request.app.state.persistence