Apna GenAI Python SDK

OpenAI-style Python SDK for the Gen AI Gateway API.

Install

pip install apna-genai

From source:

pip install -e ".[dev]"

Quickstart (sync)

from apna_genai import ApnaGenAI

client = ApnaGenAI(
    base_url="http://localhost:8080",
    api_key="tenant-token",
    scope="job_search",
)

resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

print(resp.choices[0].message.content)

Quickstart (async)

from apna_genai import AsyncApnaGenAI

async def main() -> None:
    async with AsyncApnaGenAI(
        base_url="http://localhost:8080",
        api_key="tenant-token",
        scope="job_search",
    ) as client:
        resp = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello"}],
        )
        print(resp.choices[0].message.content)

Final usage in an application

For production app code, this is the simplest setup pattern:

1. Install SDK:

pip install apna-genai

2. Configure environment once (recommended):

export APNA_GENAI_BASE_URL="https://stage-gateway.example.com"
export APNA_GENAI_API_KEY="tenant-token"
export APNA_GENAI_SCOPE="job_search"

3. Use the client in code:

from apna_genai import ApnaGenAI

client = ApnaGenAI()  # reads env vars above
resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Give me 3 interview tips"}],
)
print(resp.choices[0].message.content)

Async variant:

from apna_genai import AsyncApnaGenAI

async def main() -> None:
    async with AsyncApnaGenAI() as client:
        resp = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Give me 3 interview tips"}],
        )
        print(resp.choices[0].message.content)

Namespace layout

• Chat completions: client.chat.completions.create(...)
• Health check: client.health_check() / await client.health_check()

Environment selection (stage/prod)

The SDK does not auto-detect stage or prod. It uses this precedence for base_url:

1. explicit base_url passed to constructor
2. environment="stage"|"prod" mapped via:
   • APNA_GENAI_STAGE_BASE_URL
   • APNA_GENAI_PROD_BASE_URL
3. APNA_GENAI_BASE_URL environment variable
4. built-in default (http://localhost:8080)

Example:

export APNA_GENAI_BASE_URL="https://stage-gateway.example.com"
export APNA_GENAI_API_KEY="tenant-token"
export APNA_GENAI_SCOPE="job_search"

from apna_genai import ApnaGenAI

client = ApnaGenAI()  # picks APNA_GENAI_BASE_URL

Or with explicit environment selector:

from apna_genai import ApnaGenAI

client = ApnaGenAI(environment="stage")  # requires APNA_GENAI_STAGE_BASE_URL

Auth and required headers

• Tenant chat calls require scope and either:
  • Bearer auth (Authorization: Bearer <token>) or
  • API key header (X-API-Key: <token>) when auth_mode="x_api_key".
• Header/auth precedence is:
  1. per-request override
  2. client configuration
  3. environment fallback (APNA_GENAI_API_KEY, APNA_GENAI_SCOPE)

Per-request auth override

resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hi"}],
    api_key="override-token",
    scope="job_search",
)

Request options and retries

Use with_options(...) for scoped overrides:

fast_client = client.with_options(timeout=5.0, max_retries=5)
resp = fast_client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Ping"}],
)

Direct per-request overrides are also supported:

resp = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
    timeout=5.0,
    max_retries=3,
    headers={"X-Request-Source": "my-service"},
)

Retry policy applies to:

• network/connection errors
• timeout errors
• HTTP 408, 409, 429, and >=500

Default timeout is explicit (60s) and retries default to 2.

Error handling

from apna_genai import APIConnectionError, APIStatusError, RateLimitError

try:
    client.chat.completions.create(model="gpt-4o", messages=[{"role": "user", "content": "Hi"}])
except RateLimitError:
    print("Retry later")
except APIConnectionError:
    print("Network issue")
except APIStatusError as err:
    print(err.status_code, err.request_id, err.body)

Status-specific exceptions include: BadRequestError, AuthenticationError, PermissionDeniedError, NotFoundError, ConflictError, UnprocessableEntityError, RateLimitError, InternalServerError.

Raw response access

For advanced users who need headers and the original httpx.Response:

raw = client.chat.completions.with_raw_response.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello"}],
)

print(raw.parsed.choices[0].message.content)
print(raw.cost_quota_warning)
print(raw.http_response.status_code)

Request-level options (timeout, max_retries, headers, base_url, api_key, scope, auth_mode, provider) are applied to transport/auth only and are not included in the JSON API payload.

Model routing note

Provider selection is server-side and inferred from model prefix (for example gpt-*, gemini-*, mixtral-*). The SDK passes model through without duplicating routing logic.

Streaming caveat

The SDK includes the stream request field for compatibility, but this gateway does not currently support end-to-end SSE streaming.

API surface

• Chat completions: POST /v1/chat/completions
• Health check: GET /health-check

API contract and examples

• SDK API reference: api.md
• Runnable examples: examples/
• Environment routing example: examples/environment_routing.py