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

PackageInstallUse when
axemere-gatewaypip install axemere-gatewayDirect gateway API; any provider
axemere-gateway-openaipip install axemere-gateway-openaiYou use the openai SDK
axemere-gateway-anthropicpip install axemere-gateway-anthropicYou use the anthropic SDK
axemere-gateway-googlepip install axemere-gateway-googleYou use google-genai
axemere-gateway-langchainpip install axemere-gateway-langchainYou 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",
)
FieldTypeEnv varDefaultDescription
gateway_urlstrAXEMERE_GATEWAY_URLhttp://localhost:7080Gateway base URL
gateway_tokenstr | NoneAXEMERE_GATEWAY_TOKENNoneBearer token. Falls back to AXEMERE_WORKLOAD_TOKEN
default_providerstr | NoneAXEMERE_DEFAULT_PROVIDERNoneProvider used when none is passed to execute()
default_modelstr | NoneAXEMERE_DEFAULT_MODELNoneModel used when none is passed to execute()
workload_idstr | NoneAXEMERE_WORKLOAD_IDNoneWorkload ID for attribution and policy matching
project_idstr | NoneAXEMERE_PROJECT_IDNoneProject ID for budget controls and reporting
account_idstr | NoneAXEMERE_ACCOUNT_IDNoneAccount ID for attribution
customer_idstr | NoneAXEMERE_CUSTOMER_IDNoneCustomer ID for attribution
labelsdict[str, str](JSON string){}Arbitrary key-value labels attached to every request
provider_api_keystr | NoneAXEMERE_PROVIDER_API_KEYNoneBYOK provider key forwarded to upstream
timeoutfloatAXEMERE_TIMEOUT120HTTP 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}")
ParameterTypeDefaultDescription
providerstrProvider name ("openai", "anthropic", "google", etc.)
modelstrModel identifier
messageslist[dict]OpenAI-format message list
max_tokensint1024Maximum tokens to generate
temperaturefloat | NoneNoneSampling temperature
systemstr | NoneNoneSystem prompt (automatically separated for Anthropic)
**extra_paramsAnyAdditional 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

FieldTypeDescription
contentstrAssistant text extracted from the provider response
record_idstrGateway record ID for audit lookup
record_hashstrCryptographic hash of the record
providerstrProvider that handled the request
modelstrModel that generated the response
meteringMetering | NoneToken counts and cost
provider_responsedict | NoneRaw provider response body

StreamChunk

FieldTypeDescription
contentstrText delta (empty on final chunk)
is_finalboolTrue on the last chunk
record_idstrPopulated on the final chunk
meteringMetering | NonePopulated on the final chunk

Metering

FieldTypeDescription
cost_usdfloatTotal request cost in USD
tokens_inintInput token count
tokens_outintOutput token count
cache_hit_tokensintCached prompt tokens (Anthropic)
cache_miss_tokensintUncached prompt tokens (Anthropic)
cache_creation_tokensintCache-write tokens (Anthropic)
reasoning_tokensintReasoning tokens (o-series models)
cost_breakdownlist[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}")
ExceptionWhen raisedExtra fields
GatewayErrorBase class; network failure, non-2xx HTTP, malformed responsestatus_code, response_body
PolicyDeniedErrorRequest denied by policy (HTTP 403)reason, trace, record_id
QuotaExceededErrorBudget or quota exceeded (HTTP 429)upgrade_url, retry_after
GatewayTimeoutErrorConnector 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 the Authorization header.


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