toggleai 0.1.1

Official Python SDK for ToggleAI — Feature Flags, Remote Configs, and Logging.

toggleai-sdk (Python)

Official Python SDK for ToggleAI — Feature Flags, Remote Configs, Error Monitoring, and A/B Testing.

Website • Documentation

Features

• 🌍 Async-native — built on httpx with full asyncio support.
• ⚡ Zero-latency evaluation — evaluates flags locally in-memory after the initial fetch.
• 🔄 Real-time polling — automatic background refresh keeps configs fresh.
• 🎯 Advanced targeting — full support for targeting rules, user attributes, and percentage rollouts.
• 🐞 Structured logging — batched, async-safe log & error ingestion.
• 🧪 A/B Testing — track conversions and record exposures for running experiments.
• 🖥️ Server-side evaluation — call the backend for absolute real-time accuracy.
• 🔧 Sync-friendly — init_sync() and close_sync() for Django/Flask projects.
• 🛡️ Typed — full type hints throughout, compatible with mypy --strict.

Installation

pip install toggleai-sdk
# or with uv / poetry
uv add toggleai-sdk
poetry add toggleai-sdk

Requires Python 3.10+ (for match expressions in the evaluator).

Quick Start

Async (FastAPI, etc.)

from toggleai import ToggleAIClient, EvaluationContext

client = ToggleAIClient(
    client_id="pk_live_xxxxxxxxxxxxxxxx",
    secret="sk_live_xxxxxxxxxxxxxxxx",
)
await client.init()

# Boolean flag — evaluated locally (sub-ms)
if client.get_flag("new-checkout"):
    show_new_checkout()

# Typed flag value
color = client.get_flag_value("button-color", default="#000")

# Remote config
timeout = client.get_config("api_timeout_ms", default=5000)

# With user context (for targeting + rollout)
ctx = EvaluationContext(user_id="user_123", attributes={"plan": "premium", "country": "US"})
if client.get_flag("premium-feature", ctx):
    ...

await client.close()

Sync (Django / Flask)

from toggleai import ToggleAIClient

client = ToggleAIClient(client_id="pk_live_xxx", secret="sk_live_xxx")
client.init_sync()                         # blocks until ready

enabled = client.get_flag("dark-mode")
timeout = client.get_config("api_timeout_ms", default=5000)

client.close_sync()                        # flush logs + teardown

Feature Flags

Local evaluation (instant, cached)

# Boolean
enabled = client.get_flag("my-flag")

# Typed value (multivariate flag)
color   = client.get_flag_value("button-color", default="blue")
max_qty = client.get_flag_value("max-quantity", default=10)

# Full evaluation result with reason
result = client.evaluate_flag("my-flag", ctx)
print(result.reason)        # "TARGETING_MATCH", "ROLLOUT", "DEFAULT", …
print(result.variation_key) # None or "variation-a"

# Evaluate all flags at once
all_results = client.evaluate_all_flags(ctx)

Server-side evaluation (real-time)

# Single flag
result = await client.evaluate_flag_remote("my-flag", ctx)

# All flags
results = await client.evaluate_all_flags_remote(ctx)

Remote Configs

# Single value (typed by caller)
timeout: int = client.get_config("api_timeout_ms", default=5000)
theme:   dict = client.get_config("theme_colors", default={"primary": "#fff"})

# All configs
all_configs = client.get_all_configs()  # Dict[str, Any]

# Existence check
if client.has_config("feature_rollout_message"):
    ...

Logging & Error Monitoring

logger = client.get_logger()

logger.info("User signed in", context={"userId": "u_123"})
logger.warn("Slow query detected", duration_ms=450.0)
logger.error("Payment failed", context={"orderId": "o_99"})
logger.fatal("Out of memory")

# Capture an exception with stack trace
try:
    await risky_operation()
except Exception as exc:
    logger.capture_error(exc, context={"endpoint": "/checkout"})

# Set a default context applied to all subsequent logs
logger.set_context({"service": "payments", "version": "2.1.0"})

# Manual flush (call before shutdown)
await logger.flush()

A/B Experiment Tracking

Track a conversion event

from toggleai import TrackConversionOptions

await client.track_conversion(TrackConversionOptions(
    experiment_id="exp_xyz",
    variation_id="var_abc",   # DB UUID from evaluate_flag().variation_key
    metric_key="signup",
    value=1.0,
    user_id="user_42",
))

Auto-experiment event tracking

from toggleai import TrackEventOptions

await client.track(TrackEventOptions(
    metric_key="purchase_completed",
    user_identifier="user_42",
    value=29.99,
))

Record an exposure

from toggleai import ExposureOptions

await client.record_exposure(ExposureOptions(
    experiment_id="exp_xyz",
    variation_id="var_abc",
    user_identifier="user_42",
))

Event Listeners

client = ToggleAIClient(
    client_id="pk_live_xxx",
    secret="sk_live_xxx",
    on_ready=lambda: print("ToggleAI ready!"),
    on_config_update=lambda p: print(f"Config refreshed at {p.generated_at}"),
    on_error=lambda e: print(f"SDK error: {e.code} — {e}"),
)

Default Context

client = ToggleAIClient(
    client_id="...",
    secret="...",
    default_context=EvaluationContext(
        attributes={"server_region": "us-east-1", "app_version": "3.0.0"}
    ),
)

The default context is merged with every per-call context (per-call takes precedence).

Error Handling

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}")

Inspection

payload = client.get_payload()          # ConfigPayload | None
flag_def = client.get_flag_definition("my-flag")
flag_keys = client.get_flag_keys()      # List[str]
config_keys = client.get_config_keys()  # List[str]
env = client.get_environment()          # {"id": ..., "slug": ...}

API Reference

Client

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	Is client initialised?
get_state()	ClientState	Current lifecycle state

Feature Flags

Method	Returns	Description
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_all_flags(ctx?)	Dict[str, ...]	All flags, local
evaluate_flag_remote(key, ctx?)	Awaitable[FlagEvaluationResult]	Server-side single
evaluate_all_flags_remote(ctx?)	Awaitable[Dict[str, ...]]	Server-side all

Remote Configs

Method	Returns	Description
get_config(key, default?)	Any	Config value
get_all_configs()	Dict[str, Any]	All config values
has_config(key)	bool	Key existence check
get_config_keys()	List[str]	All config keys

Logging

Method	Description
get_logger()	Get the attached ToggleAILogger
logger.info(msg, **kw)	Log an info message
logger.error(msg_or_exc, **kw)	Log error or exception
logger.capture_error(exc, **kw)	Capture exception + stack trace
logger.set_context(ctx)	Set default log context
await logger.flush()	Flush queued events

Backend Endpoints

Endpoint	Method	Description
/sdk/config	GET	Fetch full config payload
/sdk/evaluate	POST	Server-side evaluate all flags
/sdk/evaluate/:flagKey	POST	Server-side evaluate single flag
/sdk/connect	POST	Register SDK connection
/sdk/logs/ingest	POST	Ingest a single log event
/sdk/logs/ingest/batch	POST	Ingest a batch of log events
/sdk/experiments/:id/track	POST	Track experiment conversion
/sdk/track	POST	Auto-experiment event tracking
/sdk/expose	POST	Record experiment exposures

Development

# Install dev deps
pip install -e ".[dev]"

# Run tests
pytest

# Type check
mypy toggleai

# Lint
ruff check toggleai tests