Serve the Python SDK with FastAPI

FastAPI is a natural boundary for the async-first SDK. The API layer should authenticate the caller, validate and bound the request, select provider policy, and translate SDK errors into safe HTTP responses.

Install API dependencies

pip install "zhivex-ai-sdk[api]"

Add a generation endpoint

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from zhivex_ai import (
    ConfigurationError,
    ProviderHTTPError,
    UnsupportedFeatureError,
    ValidationError,
    create_openai,
    generate_text,
)

app = FastAPI(title="Zhivex API")


class ChatRequest(BaseModel):
    prompt: str = Field(min_length=1, max_length=20_000)


@app.post("/v1/chat")
async def chat(request: ChatRequest) -> dict[str, object]:
    provider = create_openai()

    try:
        result = await generate_text(
            model=provider("gpt-5.6-terra"),
            prompt=request.prompt,
            timeout_ms=30_000,
        )
    except (ValidationError, UnsupportedFeatureError) as error:
        raise HTTPException(status_code=400, detail=str(error)) from error
    except ConfigurationError as error:
        raise HTTPException(status_code=500, detail="Server configuration error.") from error
    except ProviderHTTPError as error:
        status = 503 if error.retryable else 502
        raise HTTPException(status_code=status, detail="Upstream provider failed.") from error

    return {
        "text": result.text,
        "finish_reason": result.finish_reason,
    }

Never return raw provider response bodies, credentials, or internal configuration details to the client.

Stream text

The SDK response helper exposes an async body that FastAPI can forward:

from typing import Any

from fastapi.responses import StreamingResponse
from zhivex_ai import create_openai, stream_text, to_text_stream_response


def to_fastapi_stream(response: Any) -> StreamingResponse:
    return StreamingResponse(
        response.body,
        status_code=response.status_code,
        headers=response.headers,
        media_type=response.headers.get("content-type"),
    )


@app.post("/v1/chat/stream")
async def stream_chat(request: ChatRequest) -> StreamingResponse:
    provider = create_openai()
    result = stream_text(
        model=provider("gpt-5.6-terra"),
        prompt=request.prompt,
        timeout_ms=30_000,
    )
    return to_fastapi_stream(to_text_stream_response(result))

For UI event streams, use to_ui_message_stream_response() instead.

Harden the boundary

• authenticate before constructing a model request;
• bind credentials and model choice to a server-owned tenant policy;
• add distributed rate and concurrency limits before provider work;
• use an idempotency key when a user action can be retried;
• honor ProviderHTTPError.retry_after_ms for retryable upstream failures;
• use Postgres-backed state for durable approvals, replay, and workers;
• export redacted traces, not raw prompts by default.

The application owns billing, authorization, approval UI, retention, and retry/dead-letter policy. Continue with errors and troubleshooting, observability and evaluations, and the shared production guide.