Python SDK reference

avsb is the server-side Python SDK. Your process downloads one JSON datafile from the CDN, evaluates flags locally in microseconds, and posts exposures and conversions in batches. There is no proxy to run and no request to A vs B on the hot path.

Python 3.10 or later. The package ships type hints and a py.typed marker, so your own type checker sees every signature on this page.

Install

Shell

pip install avsb

Shell1 line

Version 1.0.1 of avsb is on its way to PyPI with this release. Until it arrives there, pip install reports that it cannot find a matching version.

Optional extras, each pulling only what it needs:

Extra	What it adds
`avsb[async]`	`AsyncAvsbServer`, on httpx
`avsb[fastapi]`	FastAPI and Starlette middleware
`avsb[flask]`	Flask extension
`avsb[django]`	Django middleware
`avsb[redis]`	Redis sticky-bucket store
`avsb[aws]`	DynamoDB sticky-bucket store
`avsb[postgres]`	PostgreSQL sticky-bucket store
`avsb[openfeature]`	OpenFeature provider

The distribution is named avsb. There is no avsb-python package.

Quickstart

Python

import osfrom avsb import AvsbServer, SingleContextserver = AvsbServer(sdk_key=os.environ["AVSB_SDK_KEY"])server.on_ready()ctx = SingleContext.user(key="u_1", plan="pro")if server.get_bool_flag("checkout_v2", False, ctx):    serve_new_checkout()server.track("checkout_completed", ctx, revenue=49.99)server.close()

Python13 lines

The context-manager form readies and closes for you:

Python

with AvsbServer(sdk_key=os.environ["AVSB_SDK_KEY"]) as server:    ...

Python2 lines

Your SDK key is a public identifier, not a secret: it is safe to ship in browser and mobile bundles, it can only fetch that environment's flag configuration and send events, and it can never read or change anything in your dashboard. Credentials covers all four A vs B credentials and which one to reach for.

Build one client per process. A client owns two background threads (a datafile refresh loop and an event delivery loop), so one per request would leak both.

Lifecycle

Python

def on_ready(self, timeout: float | None = None) -> InitResultdef refresh_datafile(self) -> booldef flush(self) -> Nonedef close(self) -> Nonedef is_ready(self) -> booldef is_polling(self) -> booldef get_init_result(self) -> InitResult | None

Python7 lines

on_ready() fetches your configuration and starts the two background threads: one refreshing every 30 seconds (jittered), one delivering events every 5 seconds. Call it once at startup and close() at shutdown, so queued events are delivered rather than lost.

It never raises. A failed fetch comes back inside the result:

Python

class InitResult:    success: bool    source: Literal["network", "bootstrap", "timeout", "error"]    error: Exception | None = None    degraded: bool = False

Python5 lines

Python

result = server.on_ready()if not result.success:    logging.error("A vs B could not start: %s", result.error)    # Flags serve your defaults, and polling keeps retrying: a blip at deploy    # time does not brick anything until the next restart.

Python5 lines

Before a configuration is loaded, every read returns the default you passed with source not_ready. That is deliberately distinct from not_found: one means the SDK has not started, the other means no such flag exists, and the fixes differ.

A certificate error on macOS

The client fetches with Python's standard library, so it trusts the certificates your Python install came with. A python.org install on macOS starts without them, and on_ready() then reports that it could not verify the CDN's certificate. Fix it once per install: run the Install Certificates.command that ships with that Python, in its /Applications/Python 3.x/ folder, or set SSL_CERT_FILE=/etc/ssl/cert.pem.

Contexts

A context is who you are evaluating: a kind, a stable key, and any attributes you target on.

Python

class SingleContext:    kind: str    key: str    name: str | None = None    attributes: dict[str, Any] = {}    @classmethod    def user(cls, key: str, **attributes: Any) -> SingleContext    @classmethod    def of(cls, kind: str, key: str, **attributes: Any) -> SingleContext

Python10 lines

Python

from avsb import SingleContextctx = SingleContext.user(key="u_123", plan="pro", beta_tester=True)org = SingleContext.of("organization", "org_acme", tier="enterprise")

Python4 lines

Bucketing is a pure function of the key, so the same key always lands in the same variation.

Multi-context

Python

from avsb import MultiContext, SingleContextctx = MultiContext.of(    SingleContext.user(key="u_123"),    SingleContext.of("organization", "org_acme", tier="enterprise"),)flag = server.get_bool_flag("enterprise_feature", False, ctx)

Python8 lines

Python

class MultiContext:    kind: Literal["multi"]    contexts: dict[str, SingleContext] = {}    @classmethod    def of(cls, *contexts: SingleContext) -> MultiContext

Python6 lines

EvalContext is the union of the two. A rule keyed on organization.key buckets every user in that organization identically, which is what makes an org-wide rollout possible.

Private attributes

Python

from avsb import ContextMeta, SingleContextctx = SingleContext(    kind="user",    key="u_123",    attributes={"plan": "pro", "email": "alice@example.com"},    _meta=ContextMeta(private_attributes=["email"]),)

Python8 lines

An attribute named in _meta.private_attributes is never written to a decision log or an exposure event.

Reading flags

Python

def get_bool_flag(self, flag_key: str, default_value: bool,                  context: EvalContext | None = None, *,                  decide_options: Iterable[DecideOption] | None = None) -> booldef get_string_flag(self, flag_key: str, default_value: str,                    context: EvalContext | None = None, *,                    decide_options: Iterable[DecideOption] | None = None) -> strdef get_number_flag(self, flag_key: str, default_value: float,                    context: EvalContext | None = None, *,                    decide_options: Iterable[DecideOption] | None = None) -> floatdef get_json_flag(self, flag_key: str, default_value: Any,                  context: EvalContext | None = None, *,                  decide_options: Iterable[DecideOption] | None = None) -> Anydef get_flag(self, flag_key: str, default_value: Any,             context: EvalContext | None = None, *,             decide_options: Iterable[DecideOption] | None = None) -> Flagdef get_all_flags(self, context: EvalContext | None = None, *,                  fire_exposures: bool = False) -> dict[str, Flag]

Python22 lines

The four typed getters return the value; get_flag returns the whole Flag envelope. A default is required on every call, so a read always has a defined answer.

Python

show_banner = server.get_bool_flag("promo_banner", False, ctx)theme       = server.get_string_flag("ui_theme", "light", ctx)max_retries = server.get_number_flag("max_retries", 3.0, ctx)config      = server.get_json_flag("api_config", {"timeout": 5000}, ctx)

Python4 lines

A type mismatch returns your default and says so. Call get_bool_flag on a flag the dashboard declares as a string and you get your default plus a warning naming the right getter. Nothing is coerced: bool("false") is True in Python, and shipping that silently was a real bug.

The `Flag` envelope

Python

@dataclass(frozen=True)class Flag:    value: Any    variation_key: str | None    source: EvaluationSource    rule_id: str | None    rule_type: RuleType | None    reasons: tuple[str, ...]    evaluated_at: int    # ms since the Unix epoch    duration_micros: int    def is_enabled(self) -> bool    def exists(self) -> bool

Python13 lines

reasons is the ordered decision trail, always collected, and the fastest way to answer "why did this user get that?":

Plain text

rule "rule_paid_users" skipped (audience not matched)rule "rule_ab_split" matched (ab_test), bucketed into variation "treatment" (hash 4821/5000)

Plain text2 lines

Every `source` value

Python

EvaluationSource = Literal[    "datafileOverride", "runtimeOverride", "sticky", "rule", "holdout",    "bandit", "default", "not_found", "disabled", "not_ready",]RuleType = Literal["targeted_delivery", "ab_test", "holdout", "bandit"]

Python6 lines

A decision produced the value:

Value	Meaning
`datafileOverride`	An override pinned to this context in the dashboard.
`runtimeOverride`	An override set in code with `set_override`.
`sticky`	A stored assignment replayed from your sticky-bucket store.
`rule`	A targeting rule matched. `rule_type` says which kind.
`holdout`	Held out for measurement, so the flag stays at its default variation.
`bandit`	A bandit rule chose the variation.

A fallback produced the value, and is_enabled() is always False:

Value	Meaning	What to do
`default`	The flag exists and no rule matched.	Nothing, this is normal.
`not_found`	No flag with that key in this environment.	Check the key, and that it is published.
`disabled`	The flag is turned off for this environment.	Turn it on in the dashboard.
`not_ready`	No configuration loaded yet, or the client was closed.	Call `on_ready()` at startup.

is_enabled() is True when a real decision produced the value and the value is truthy. Truthiness follows JavaScript's rules, identically in every A vs B SDK, so a flag reads the same from Python, Node, Go and the rest: falsy is False, 0, "", None, NaN; everything else is truthy, including [] and {}, which Python itself would call falsy.

exists() is False only for not_found and not_ready. A flag you turned off still exists.

Reading without recording an exposure

Python

from avsb import DecideOptionflag = server.get_flag(    "checkout_v2", False, ctx,    decide_options=[DecideOption.DISABLE_EXPOSURE],)

Python6 lines

Python

class DecideOption(str, Enum):    DISABLE_EXPOSURE = "DISABLE_EXPOSURE"    INCLUDE_REASONS = "INCLUDE_REASONS"   # accepted for symmetry; reasons are always on

Python3 lines

get_all_flags suppresses exposures by default, so rendering a debug table does not enrol the viewer in every experiment at once.

Request-scoped evaluation

Python

def for_user(self, context: EvalContext) -> UserBoundClient

Python1 line

Python

avsb = server.for_user(SingleContext.user(key=user_id, plan=plan))if avsb.get_bool_flag("checkout_v2", False):    ...avsb.track("checkout_started")

Python5 lines

UserBoundClient mirrors every getter plus get_all_flags, track and manual_exposure, all without the context argument. It is a binding, not a second client: nothing to close, and a configuration refresh is visible immediately. Purchases stay on the client itself, because an order sends immediately and its shape differs between the sync and async clients.

Recording events

Python

def track(self, event_key: str, context: EvalContext | None = None, *,          revenue: float | None = None,          value: float | None = None,          properties: Mapping[str, Any] | None = None) -> Nonedef track_purchase(self, visitor_id: str, order: PurchaseOrder) -> Nonedef manual_exposure(self, flag_key: str, context: EvalContext) -> None

Python8 lines

Python

server.track("signup_completed", ctx)server.track("checkout_completed", ctx, revenue=49.99)server.track("items_in_basket", ctx, value=3)

Python3 lines

revenue is money, in decimal major units of the project currency.
value is the number behind an average-value metric. It is a separate column, never money.
properties is accepted for symmetry with exposures, but conversion events have no properties field, so it cannot be stored. Passing it logs one warning naming the drop rather than pretending it arrived.

Attribution happens server-side: A vs B links a conversion to the experiments that visitor was exposed to, so you never pass an experiment id.

Python

from avsb import OrderItem, PurchaseOrderserver.track_purchase(    "u_123",    PurchaseOrder(        order_id="ORD-1001",        total=99.95,        currency="USD",        items=[OrderItem(sku="SKU-1", price=99.95, quantity=1)],    ),)

Python11 lines

Orders send immediately rather than waiting for the batch timer, because an order is the event you least want buffered. Exposures are recorded automatically when a rule, bandit, holdout or sticky assignment decides a flag; call manual_exposure only when you read a flag early and the value reached a person later.

Events are batched, retried on a jittered backoff curve, and flushed on close(). Every event carries an id stamped once at enqueue time, so a retried batch is deduplicated rather than double-counted. A full queue drops the oldest events and logs how many.

The async client

Shell

pip install "avsb[async]"

Shell1 line

Python

import osfrom avsb import SingleContextfrom avsb.async_client import AsyncAvsbServerasync def main() -> None:    async with AsyncAvsbServer(sdk_key=os.environ["AVSB_SDK_KEY"]) as server:        ctx = SingleContext.user(key="u_1", plan="pro")        if server.get_bool_flag("checkout_v2", False, ctx):            ...        await server.track("checkout_started", ctx)

Python10 lines

Awaited: on_ready, refresh_datafile, track, track_purchase, flush, close.

Not awaited: every getter, get_all_flags, for_user, manual_exposure, and the override methods. Evaluation is a dict lookup and a hash against an in-memory configuration with no I/O, so making it a coroutine would add an event-loop round trip per read and buy nothing. It also means ordinary synchronous helpers inside an async handler can read flags.

The one mistake people make with this client: dropping await on track() or track_purchase(). Python does not run any of an async def function's body until something awaits it, so server.track("checkout_completed", ctx) without await builds a coroutine and throws it away. Nothing is queued, no error is raised at the call site, and the only clue is a background "coroutine was never awaited" warning, often logged far from where you called it.

Everything else matches AvsbServer name for name. Pass http_client= to share your application's httpx client; a client you pass is left open by close(), one the SDK opened is closed.

Framework integrations

Framework	Import	Guide
FastAPI, Starlette	`from avsb.middleware.fastapi import AvsbFastApiMiddleware`	FastAPI (and Flask)
Flask	`from avsb.middleware.flask import AvsbFlask`	FastAPI (and Flask)
Django	`from avsb.middleware.django import AvsbDjangoMiddleware`	Django
Any ASGI app	`from avsb.middleware.asgi import AvsbAsgiMiddleware`	This page
Any WSGI app	`from avsb.middleware.wsgi import AvsbWsgiMiddleware`	This page

AvsbFastApiMiddleware is the ASGI middleware under its FastAPI-facing name, so the two are the same class. Each middleware binds a per-request client from the context you build, so handlers read flags without passing a context around.

Sticky bucketing

Python

import osimport redisfrom avsb import AvsbServerfrom avsb.sticky.redis import RedisStickyBucketServiceserver = AvsbServer(    sdk_key=os.environ["AVSB_SDK_KEY"],    sticky_bucket_service=RedisStickyBucketService(redis.Redis()),)

Python9 lines

Also available: avsb.sticky.memory.InMemoryStickyBucketService, avsb.sticky.dynamodb.DynamoDBStickyBucketService, and avsb.sticky.postgres.PostgresStickyBucketService. Assignments are saved for A/B tests only: a rollout is not an assignment, and a holdout is not a decision worth remembering. A store outage degrades bucketing stability and logs it; it never breaks an evaluation.

Bring your own with two methods:

Python

class StickyBucketService(Protocol):    def lookup(self, user_id: str, flag_key: str) -> StickyAssignment | None: ...    def save(self, user_id: str, flag_key: str, assignment: StickyAssignment) -> None: ...

Python3 lines

Overrides and the decision log

Python

server.set_override("checkout_v2", "treatment")server.set_override_for_context("checkout_v2", "u_123", "control")server.clear_override("checkout_v2")server.clear_all_overrides()

Python4 lines

An override is applied ahead of everything else and reported with source runtimeOverride. It lives in the process, so it disappears on restart and does not exist for any other instance: local development and incident response, not configuration.

Python

log = server.flush_decision_log()for entry in log.entries:    print(entry.flag_key, entry.source, entry.value)if log.dropped:    print(f"{log.dropped} entries were dropped before this flush")

Python5 lines

Every evaluation is recorded with its reasons and its context summary (kinds and keys only, never attribute values), and any value that touched a private attribute is replaced with [REDACTED]. The collector keeps the most recent 10,000 entries and says so when it starts dropping; size it with decision_log_limit.

Logging

The default logger prints warnings to stderr in development and stays silent in production, where your application's logging configuration governs. Development is anything whose AVSB_ENV, ENVIRONMENT, ENV, PYTHON_ENV, APP_ENV, FLASK_ENV or DJANGO_ENV is not production, and an unconfigured process counts as development.

Shell

export AVSB_LOG_LEVEL=debug   # or info, warn, error, none

Shell1 line

Python

from avsb import AvsbLogger, NoopLoggerserver = AvsbServer(sdk_key=..., logger=AvsbLogger("myapp.flags"))server = AvsbServer(sdk_key=..., logger=NoopLogger())

Python4 lines

What the SDK tells you out loud: an SDK key that does not look like one (naming what it looks like instead), a configuration fetch failure with the URL, the status and a curl to reproduce, evaluating with no context (once per process), reading a flag with the wrong typed getter, a full event queue or decision log with the count dropped, and events lost after every retry.

Testing

Shell

pip install "avsb[test]"

Shell1 line

Python

import pytestfrom avsb import AvsbServer, SingleContextfrom avsb.test_utils import build_boolean_flag_entry, build_minimal_datafile@pytest.fixture()def server():    datafile = build_minimal_datafile(        flags=[build_boolean_flag_entry("checkout_v2", default_value=True)]    )    client = AvsbServer(        "sdk_production_xxxxxxxxxxxxxxxx",        datafile=datafile,        polling_enabled=False,        events_enabled=False,    )    yield client    client.close()def test_new_checkout_is_on(server):    assert server.get_bool_flag("checkout_v2", False, SingleContext.user(key="u_1"))

Python20 lines

polling_enabled=False and events_enabled=False keep the run off the network entirely. avsb.test_utils also ships build_ab_test_rule, build_targeted_delivery_rule, InMemoryDatafileLoader, FakeClock and TEST_SDK_KEY.

The full constructor

Python

AvsbServer(    sdk_key: str,    *,    datafile: FlagDatafile | Mapping[str, Any] | None = None,    datafile_url: str | None = None,    cdn_host: str | None = None,    polling_interval_seconds: float = 30.0,    polling_enabled: bool = True,    init_timeout_seconds: float = 10.0,    events_enabled: bool = True,    tracker_options: TrackerOptions | None = None,    sticky_bucket_service: StickyBucketService | None = None,    bandit_model_store: BanditModelStore | None = None,    runtime_overrides: Mapping[str, str] | None = None,    decision_log_limit: int | None = None,    logger: Logger | None = None,    transport: DatafileTransport | None = None,    event_sender: EventSender | None = None,    heartbeat_sender: HeartbeatSender | None = None,)

Python20 lines

AsyncAvsbServer takes the same parameters minus transport, event_sender and heartbeat_sender, plus http_client: httpx.AsyncClient | None.

Polling is floored at 5 seconds and jittered by plus or minus 20 percent, so a fleet that deployed together does not fetch in lockstep.

Django and FastAPI (and Flask): request-scoped setup per framework.
Multi-context identity
Sticky bucketing
Credentials: all four A vs B credentials and which one to reach for.