Axemere Gateway — Python SDK
Add governance, cost controls, and an audit trail to every AI call, without changing your application code. The Python SDK is a drop-in replacement for the OpenAI, Anthropic, Google Gemini, and LangChain clients.
GitHub → Axemere-LLC/axemere-python
Packages
| Package | Install | Use when |
|---|---|---|
axemere-gateway | pip install axemere-gateway | Direct gateway API; any provider |
axemere-gateway-openai | pip install axemere-gateway-openai | You use the openai SDK |
axemere-gateway-anthropic | pip install axemere-gateway-anthropic | You use the anthropic SDK |
axemere-gateway-google | pip install axemere-gateway-google | You use google-genai |
axemere-gateway-langchain | pip install axemere-gateway-langchain | You use LangChain |
Requires Python 3.10+.
Quick start
pip install axemere-gateway-openai
from axemere.gateway.openai import OpenAI # reads AXEMERE_GATEWAY_URL and AXEMERE_GATEWAY_TOKEN from env client = OpenAI() response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello"}], ) print(response.choices[0].message.content)
No other code changes required. Every request is governed, costed, and recorded by the gateway.
Configuration
AiGatewayConfig
from axemere.gateway import AiGatewayConfig
All fields read from environment variables at construction time. Pass keyword arguments to override individual values.
config = AiGatewayConfig( gateway_url="http://localhost:7080", gateway_token="tok_...", workload_id="wl_my_app", )
| Field | Type | Env var | Default | Description |
|---|---|---|---|---|
gateway_url | str | AXEMERE_GATEWAY_URL | http://localhost:7080 | Gateway base URL |
gateway_token | str | None | AXEMERE_GATEWAY_TOKEN | None | Bearer token. Falls back to AXEMERE_WORKLOAD_TOKEN |
default_provider | str | None | AXEMERE_DEFAULT_PROVIDER | None | Provider used when none is passed to execute() |
default_model | str | None | AXEMERE_DEFAULT_MODEL | None | Model used when none is passed to execute() |
workload_id | str | None | AXEMERE_WORKLOAD_ID | None | Workload ID for attribution and policy matching |
project_id | str | None | AXEMERE_PROJECT_ID | None | Project ID for budget controls and reporting |
account_id | str | None | AXEMERE_ACCOUNT_ID | None | Account ID for attribution |
customer_id | str | None | AXEMERE_CUSTOMER_ID | None | Customer ID for attribution |
labels | dict[str, str] | (JSON string) | {} | Arbitrary key-value labels attached to every request |
provider_api_key | str | None | AXEMERE_PROVIDER_API_KEY | None | BYOK provider key forwarded to upstream |
timeout | float | AXEMERE_TIMEOUT | 120 | HTTP timeout in seconds |
Methods
AiGatewayConfig.from_env() → AiGatewayConfig
Class method. Equivalent to AiGatewayConfig(), provided for explicitness.
config.set_defaults(provider: str, model: str) → None
Mutates default_provider and default_model in place. Useful when switching providers inside a loop.
config.proxy_url(provider: str) → str
Returns the proxy base URL for a given provider. Use this as the base_url for an existing provider SDK (see Proxy mode).
AiGatewayClient
from axemere.gateway import AiGatewayClient
Framework-independent async/sync client for the gateway's explicit action API. Use this when you want full control or are not using an OpenAI/Anthropic/Google SDK.
from axemere.gateway import AiGatewayClient, AiGatewayConfig client = AiGatewayClient(AiGatewayConfig())
await client.execute(...) → ExecuteResponse
response = await client.execute( provider="openai", model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello"}], max_tokens=512, ) print(response.content) print(f"Cost: ${response.metering.cost_usd:.6f}")
| Parameter | Type | Default | Description |
|---|---|---|---|
provider | str | — | Provider name ("openai", "anthropic", "google", etc.) |
model | str | — | Model identifier |
messages | list[dict] | — | OpenAI-format message list |
max_tokens | int | 1024 | Maximum tokens to generate |
temperature | float | None | None | Sampling temperature |
system | str | None | None | System prompt (automatically separated for Anthropic) |
**extra_params | Any | — | Additional parameters forwarded to the provider |
client.execute_sync(...) → ExecuteResponse
Synchronous wrapper around execute(). Do not call inside an already-running async context.
await client.stream(...) → AsyncIterator[StreamChunk]
async for chunk in await client.stream( provider="anthropic", model="claude-haiku-4-5-20251001", messages=[{"role": "user", "content": "Count to 5"}], ): if not chunk.is_final: print(chunk.content, end="", flush=True) else: print(f"\nTokens: {chunk.metering.tokens_in} in / {chunk.metering.tokens_out} out")
Accepts the same parameters as execute(). Yields StreamChunk objects; the final chunk has is_final=True and carries metering and record_id.
client.stream_sync(...) → list[StreamChunk]
Synchronous wrapper. Collects all chunks and returns them as a list.
Response types
ExecuteResponse
| Field | Type | Description |
|---|---|---|
content | str | Assistant text extracted from the provider response |
record_id | str | Gateway record ID for audit lookup |
record_hash | str | Cryptographic hash of the record |
provider | str | Provider that handled the request |
model | str | Model that generated the response |
metering | Metering | None | Token counts and cost |
provider_response | dict | None | Raw provider response body |
StreamChunk
| Field | Type | Description |
|---|---|---|
content | str | Text delta (empty on final chunk) |
is_final | bool | True on the last chunk |
record_id | str | Populated on the final chunk |
metering | Metering | None | Populated on the final chunk |
Metering
| Field | Type | Description |
|---|---|---|
cost_usd | float | Total request cost in USD |
tokens_in | int | Input token count |
tokens_out | int | Output token count |
cache_hit_tokens | int | Cached prompt tokens (Anthropic) |
cache_miss_tokens | int | Uncached prompt tokens (Anthropic) |
cache_creation_tokens | int | Cache-write tokens (Anthropic) |
reasoning_tokens | int | Reasoning tokens (o-series models) |
cost_breakdown | list[CostBreakdownItem] | Per-tier cost line items |
Errors
All errors inherit from GatewayError.
from axemere.gateway import GatewayError, PolicyDeniedError, QuotaExceededError, GatewayTimeoutError try: response = await client.execute(...) except PolicyDeniedError as e: print(f"Denied: {e.reason}") # human-readable denial reason print(f"Record: {e.record_id}") except QuotaExceededError as e: print(f"Quota exceeded. Upgrade at: {e.upgrade_url}") except GatewayTimeoutError: print("Gateway connector timed out") except GatewayError as e: print(f"Gateway error (HTTP {e.status_code}): {e}")
| Exception | When raised | Extra fields |
|---|---|---|
GatewayError | Base class; network failure, non-2xx HTTP, malformed response | status_code, response_body |
PolicyDeniedError | Request denied by policy (HTTP 403) | reason, trace, record_id |
QuotaExceededError | Budget or quota exceeded (HTTP 429) | upgrade_url, retry_after |
GatewayTimeoutError | Connector timed out (HTTP 504) | — |
Wrapper packages
The four wrapper packages are drop-in replacements for their respective provider SDKs. Swap the import and you're done: no other code changes required.
axemere-gateway-openai
# Before from openai import OpenAI, AsyncOpenAI client = OpenAI() # After from axemere.gateway.openai import OpenAI, AsyncOpenAI client = OpenAI() # reads AXEMERE_* env vars automatically
Also exports AzureOpenAI and AsyncAzureOpenAI. Azure endpoints are auto-detected from the azure_endpoint or AZURE_OPENAI_ENDPOINT env var.
axemere-gateway-anthropic
# Before from anthropic import Anthropic, AsyncAnthropic client = Anthropic() # After from axemere.gateway.anthropic import Anthropic, AsyncAnthropic client = Anthropic()
axemere-gateway-google
# Before import google.generativeai as genai genai.configure(api_key="...") # After from axemere.gateway.google import genai_client genai = genai_client() # returns a configured google.generativeai module-like object model = genai.GenerativeModel("gemini-1.5-flash")
axemere-gateway-langchain
Provides ChatOpenAI and ChatAnthropic subclasses pre-configured to route through the gateway, plus convenience re-exports for proxy-mode use.
from axemere.gateway.langchain import ChatOpenAI, ChatAnthropic llm = ChatOpenAI(model="gpt-4o-mini") response = llm.invoke("Explain monads in one sentence.") print(response.content)
Full example → Axemere-LLC/langchain-gateway-demo — a multi-agent reference implementation built with LangChain and the Axemere Gateway.
Proxy mode
Use proxy_url() to point any existing provider SDK at the gateway without switching packages. The gateway intercepts the standard API call, applies policy, and forwards it upstream.
import openai from axemere.gateway import AiGatewayConfig cfg = AiGatewayConfig() client = openai.OpenAI( base_url=cfg.proxy_url("openai"), api_key="ignored", # gateway handles auth ) response = client.chat.completions.create( model="gpt-4o-mini", messages=[{"role": "user", "content": "Hello"}], )
Note: The gateway token is embedded in the proxy URL path (
/k/{token}). URLs appear in access logs and browser history. Prefer the explicit action API (AiGatewayClient.execute()) or the wrapper packages: they send the token in theAuthorizationheader.
Get a gateway
- Free tier — self-hosted, no account required → Install
- Self-Hosted — team features + control plane → Get started
- Managed — fully hosted, zero-ops → Get started