Python
Async-native Python SDK built on httpx. Full support for FastAPI, Django, Flask, and any Python 3.10+ application with both async and sync modes.
pip install toggleai-sdkThe Python SDK is designed from the ground up for high-concurrency async runtimes and synchronous frameworks alike. It works in close synergy with ToggleAI backend services to guarantee maximum speed, safety, and background data synchronization:
⚡ Local In-Memory Evaluation
Upon calling await client.init() (or client.init_sync() for synchronous runtimes), the SDK retrieves the complete flag ruleset from GET /sdk/config. Subsequent resolutions with client.get_flag() or client.get_flag_value() are computed instantly in memory under sub-millisecond speeds, completely removing HTTP bottlenecks from your critical application paths.
🌐 Real-Time Remote Evaluation
For sensitive operations demanding absolute fresh synchronization or immediate validation bypassing local caches, client.evaluate_flag_remote() triggers a direct, secure edge network call to POST /sdk/evaluate. This guarantees exact evaluation logic in real-time.
import asyncio
from toggleai import ToggleAIClient, EvaluationContext
async def main():
# Instantiation
client = ToggleAIClient(
client_id="pk_live_xxx",
secret="sk_live_xxx",
polling_interval=30000,
on_ready=lambda: print("ToggleAI ready!"),
on_error=lambda err: print(f"SDK Error: {err}"),
)
# Initialize the client asynchronously (starts config polling)
await client.init()
# Define user evaluation context
ctx = EvaluationContext(
user_id="user_123",
attributes={"plan": "premium", "region": "us-west"}
)
# 1. Feature Flag Evaluation (computed instantly in-memory via local cache)
if client.get_flag("new-checkout", ctx):
print("Show new checkout UI")
# 2. Remote Configuration (computed instantly in-memory via local cache)
timeout = client.get_config("api_timeout_ms", default=5000)
print(f"API Timeout: {timeout}ms")
# 3. Real-Time Remote Evaluation (direct API call, bypassing local cache)
remote_result = await client.evaluate_flag_remote("new-checkout", ctx)
print(f"Remote evaluation result: {remote_result.enabled} (reason: {remote_result.reason})")
# 4. Asynchronous Non-Blocking Logging
logger = client.get_logger()
logger.info("Checkout viewed", user_id="user_123")
# Gracefully shut down the client, flushing any queued log events and stopping the polling loop
await client.close()
if __name__ == "__main__":
asyncio.run(main())from contextlib import asynccontextmanager
from fastapi import FastAPI, Depends
from toggleai import ToggleAIClient, EvaluationContext
client = ToggleAIClient(
client_id="pk_live_xxx",
secret="sk_live_xxx",
on_ready=lambda: print("ToggleAI ready!"),
)
@asynccontextmanager
async def lifespan(app: FastAPI):
await client.init()
yield
await client.close()
app = FastAPI(lifespan=lifespan)
@app.get("/checkout")
async def checkout(user_id: str):
ctx = EvaluationContext(user_id=user_id, attributes={"plan": "premium"})
if client.get_flag("new-checkout", ctx):
return {"ui": "new"}
timeout = client.get_config("api_timeout_ms", default=5000)
return {"ui": "legacy", "timeout": timeout}Django uses synchronous views by default. Use init_sync() and close_sync().
# myapp/toggleai_config.py
from toggleai import ToggleAIClient
from django.conf import settings
client = ToggleAIClient(
client_id=settings.TOGGLEAI_CLIENT_ID,
secret=settings.TOGGLEAI_SECRET,
)
client.init_sync() # Blocks until ready# myapp/views.py
from django.http import JsonResponse
from .toggleai_config import client
def checkout_view(request):
user_id = request.user.id
if client.get_flag("new-checkout", {"userId": str(user_id)}):
return JsonResponse({"ui": "new"})
return JsonResponse({"ui": "legacy"})# myapp/apps.py — Initialize on Django startup
from django.apps import AppConfig
class MyAppConfig(AppConfig):
name = "myapp"
def ready(self):
from .toggleai_config import client # triggers init_sync()import atexit
from flask import Flask, g
from toggleai import ToggleAIClient
app = Flask(__name__)
client = ToggleAIClient(
client_id="pk_live_xxx",
secret="sk_live_xxx",
)
client.init_sync()
# Flush logs on shutdown
atexit.register(client.close_sync)
@app.route("/api/flags/<flag_key>")
def get_flag(flag_key):
enabled = client.get_flag(flag_key)
return {"key": flag_key, "enabled": enabled}from toggleai import ToggleAIError
try:
await client.init()
except ToggleAIError as exc:
match exc.code:
case "INVALID_KEY": print("Check your API credentials.")
case "RATE_LIMITED": print("Slow down.")
case "FORBIDDEN": print("Insufficient API key scope.")
case "NETWORK_ERROR": print("Cannot reach the ToggleAI API.")
case _: print(f"Unexpected error: {exc}")| Option | Type | Default |
|---|---|---|
client_id | str | required |
secret | str | required |
base_url | str | hosted URL |
polling_interval | int (ms) | 30000 |
evaluation_mode | "local" | "server" | "local" |
disable_cache | bool | False |
default_context | EvaluationContext | None |
timeout | int (ms) | 10000 |
on_ready | Callable | None |
on_config_update | Callable | None |
on_error | Callable | None |
Client Methods
| Method | Returns | Description |
|---|---|---|
init() | Awaitable[None] | Fetch config, start polling |
init_sync() | None | Sync wrapper for init() |
close() | Awaitable[None] | Stop polling, flush, teardown |
close_sync() | None | Sync wrapper for close() |
refresh() | Awaitable[None] | Manually refresh config |
is_ready() | bool | Client initialized? |
get_flag(key, ctx?, default?) | bool | Boolean flag value |
get_flag_value(key, ctx?, default?) | Any | Typed flag value |
evaluate_flag(key, ctx?) | FlagEvaluationResult | Full local evaluation |
evaluate_flag_remote(key, ctx?) | Awaitable[...] | Server-side evaluation |
get_config(key, default?) | Any | Config value |
get_all_configs() | Dict[str, Any] | All config values |
has_config(key) | bool | Key existence check |
get_logger() | ToggleAILogger | Get attached logger |
track_conversion(...) | Awaitable[None] | Track experiment conversion |
record_exposure(...) | Awaitable[None] | Record exposure |
track(...) | Awaitable[None] | Auto-experiment event |