Catalyst Python SDK

A feature flag client for the Catalyst platform that reads the current snapshot as it evaluates and decides locally.

Each check is a conditional request, so an unchanged environment costs a 304 with no body, and a toggle you flip in the dashboard is picked up on the next check. The decision itself never leaves the process. The client talks to the hosted API by default, so a working setup needs only a key and a project id.

from catalyst_sdk import CatalystClient

client = CatalystClient(
    sdk_key="cp_prod_a1b2c3d4e5f6g7h8i9j0k1",   # API Keys tab in the dashboard
    project_id="7c9e6679-7425-40de-944b-e07fc1f90ae7",
    env="prod",
)

if client.is_enabled("new-checkout", user_id="user_123",
                     attributes={"email": "alice@acme.com"}):
    ...

Repeating that identity on every check is tedious, so it can be attached once instead — on the client for a single-user script or job, or per user with for_user for a process serving many:

client = CatalystClient(sdk_key=..., project_id=..., user_id="user_123",
                        attributes={"email": "alice@acme.com"})
client.is_enabled("new-checkout")            # identity comes from the client

alice = client.for_user("alice", {"email": "alice@acme.com"})
alice.is_enabled("new-checkout")             # identity comes from the view

Install

pip install sdk-catalyst

The distribution is named sdk-catalyst (the catalyst-sdk name is taken on PyPI by an unrelated project), but the import name is unchanged:

from catalyst_sdk import CatalystClient

From a checkout:

uv sync            # or: pip install -e ".[dev]"

Only runtime dependency is httpx. Requires Python 3.11+.

Configuration

project_id is worth calling out: /api/v1/bootstrap takes it as a query parameter, so the SDK cannot infer it from the key.

Choosing the API host

# Staging, a self-hosted instance, or a tunnel to your laptop.
client = CatalystClient(..., host="https://catalyst-api-uakz.onrender.com")

Freshness

By default every check reads, so there is no propagation delay to tune. Two knobs matter when that is not what you want:

# Evaluate from memory only; drive freshness yourself.
client = CatalystClient(..., refresh_on_evaluate=False)
client.start_auto_refresh(interval=30)

# A dead API should not add its timeout to every check.
client = CatalystClient(..., failure_backoff=15.0)

• Concurrent checks collapse into a single read, so a thread pool does not stampede the API.
• After a failure, reads are held off for failure_backoff seconds, so an unreachable API costs one timeout per window rather than one per request.
• get_all() reads once regardless of how many flags it holds.

How it works

   is_enabled("new-checkout", user_id="u1")
        │
        ├──▶  GET /api/v1/bootstrap
        │      If-None-Match: <last etag>
        │        304            → snapshot unchanged, no body
        │        200            → replace snapshot
        │
        └──▶  local decision against the snapshot  →  bool

• No startup fetch. Constructing a client performs no I/O, so it cannot fail on a cold or unreachable API. The first check is what reads.
• Conditional read. Every check sends If-None-Match, so an unchanged environment costs a 304 with no body. The ETag derives from the project's environment version, which every snapshot-affecting mutation bumps.
• Atomic swap. A single attribute assignment replaces the snapshot, so evaluation never sees a half-updated view.
• Cheap on the server. The API answers from a Redis snapshot keyed by environment version, so the read does not re-query every flag, state and rule. Redis being unavailable only makes it slower.
• Fails closed, never throws. A failed read serves false. Serving a stale targeting decision is worse than not serving the feature, so there is no cached fallback anywhere in the SDK. An explicit refresh() still raises AuthorizationError, because a rejected key will not fix itself.

Evaluation precedence

The SDK mirrors the server exactly, in this order:

1. Emergency kill switch → everyone gets false, nothing else is looked at
2. Enable to all users → everyone gets true, no targeting at all
3. Targeting rules, ascending priority, first match wins → puts the user in the population. If rules exist and none matched, the user is filtered out and gets false.
4. Percentage rollout → a deterministic Murmur3 bucket over flag_key:user_id splits the eligible population, so 0% serves nobody and 100% serves everybody. A matched user outside the bucket gets the opposite of the rule's value, so the rules and the percentage partition the matched group exactly.

The bucket is a fixed number per user, so raising a percentage only ever adds users; it can never remove one. That is what makes a gradual rollout safe.

Reasons are reported on EvaluationResult.reason: KILL_SWITCH_ACTIVE, ENABLE_ALL_USERS, RULE_AND_ROLLOUT, RULE_OUTSIDE_ROLLOUT, PERCENTAGE_ROLLOUT, PERCENTAGE_OUTSIDE_ROLLOUT, DEFAULT_VALUE (rules filtered the user out), FLAG_NOT_FOUND, NO_SNAPSHOT.

Supported operators: equals, not_equals, in, not_in, contains, starts_with, ends_with, greater_than, greater_than_or_equal, less_than, less_than_or_equal, exists, not_exists. Short aliases (eq, neq, gt, gte, lt, lte, notExists) are accepted and normalized.

A missing context attribute makes a condition false, so the rule safely skips instead of raising.

API

Attaching the user

Two ways, depending on how many users one client serves.

On the client, when it always acts for the same person — a script, a cron job, a service acting on behalf of one account:

client = CatalystClient(..., user_id="user_123",
                        attributes={"email": "alice@acme.com"})
client.is_enabled("new-checkout")   # no identity repeated at the call site

Per user, when one long-lived client serves many — a web app:

sdk = CatalystClient(sdk_key=..., project_id=...)

def checkout(request):
    alice = sdk.for_user(request.user.id, {"email": request.user.email})
    if alice.is_enabled("new-checkout"):
        ...

for_user returns a UserScopedClient: a view holding no transport, snapshot, or refresh thread of its own. It shares the parent's, so scoping per request costs nothing and all views share one snapshot and one conditional read.

Rules for combining context, applied identically on both:

• A per-call user_id replaces the bound one; a per-call attributes is merged onto the bound ones, so adding one attribute never silently drops the others.
• for_user inherits the client's constructor defaults and merges on top.
• UserScopedClient has no for_user of its own, so a scoped view cannot re-bind and carry the previous user's attributes over to the next one.
• Unbound clients behave exactly as before: an empty user_id is hashed like any other value.

is_enabled(flag_key, user_id="", attributes=None) -> bool

The hot path: one conditional read, then a local decision. user_id and attributes fall back to the client's.

evaluate(flag_key, ...) -> EvaluationResult

Same evaluation, but returns the full decision:

result = client.evaluate("new-checkout", user_id="user_123",
                         attributes={"email": "alice@acme.com"})
result.value        # True
result.reason       # 'RULE_AND_ROLLOUT' | 'PERCENTAGE_ROLLOUT'
                    # | 'KILL_SWITCH_ACTIVE' | 'ENABLE_ALL_USERS'
                    # | 'DEFAULT_VALUE' | 'FLAG_NOT_FOUND' | 'NO_SNAPSHOT'
result.rule_id      # the matched rule, when a rule fired
result.flag_version # snapshot version the decision came from

get_all(user_id="", attributes=None) -> dict[str, bool]

Evaluates the whole snapshot in one pass, for rendering a UI. Also falls back to the client's context.

refresh() -> bool

Conditionally re-fetches. Returns True if a new snapshot was applied, False if the server replied 304 Not Modified.

start_auto_refresh(interval=30.0, on_error=None) -> Thread

Daemon thread that calls refresh() on an interval. on_error receives every refresh failure, including ones that were safely absorbed. Only needed with refresh_on_evaluate=False; otherwise it is redundant with the per-check read.

client.start_auto_refresh(interval=15, on_error=lambda e: log.warning(e))
...
client.stop_auto_refresh()

Health and introspection

client.is_ready      # bool  - a snapshot is loaded
client.version       # int   - environment cache version
client.flag_keys()   # list[str]  (reads memory; read after a check or refresh)
client.stats         # dict  - refresh counters, etag, host, last error
client.last_error    # str | None

Lifecycle

CatalystClient is a context manager and closes its HTTP pool:

with CatalystClient(...) as client:
    ...

Failing closed

There is no disk cache. If the SDK cannot reach Catalyst it serves false: a flag check inside someone else's request must not become a 500, and it must not serve a targeting decision the API can no longer confirm.

A 304 Not Modified is a successful read — the in-memory snapshot stays valid and keeps serving. Only a genuine failure flips to false, and the next successful read clears it.

client = CatalystClient(..., offline=True)   # no network at all; serves false

Error handling

Evaluation never raises. An unknown flag key, a missing snapshot, and a failed read all resolve to false — a flag check inside someone else's request must not become a 500, and must not turn a feature on by accident. Set raise_on_error=True when you would rather find out.

Demo

examples/fastapi_demo/app.py is a runnable FastAPI service that gates a checkout endpoint and explains its decisions.

export CATALYST_SDK_KEY="cp_prod_..."
export CATALYST_PROJECT_ID="<project uuid>"
export CATALYST_ENV="prod"
# Only when running against your own API:
# export CATALYST_HOST="http://localhost:8000"

python examples/fastapi_demo/app.py

curl localhost:9000/api/checkout?user_id=user_123 -H 'x-user-email: alice@acme.com'
curl localhost:9000/api/inspect?flag=new-checkout\&user_id=user_123
curl localhost:9000/health

Tests

pytest

Two layers guard correctness:

• The SDK's own suite covers hashing, the operator matrix, precedence, ETag handling, refresh, auto-refresh, and failing closed.

• backend/tests/test_sdk_parity.py is a differential test that runs the server's evaluator and the SDK's evaluator over the same fuzzed inputs and requires identical values, reasons, and rule ids. It also checks Murmur3 against the real mmh3 library. Run it with the backend suite:

  cd backend && uv run pytest tests/test_sdk_parity.py
  

Releasing

Releases are automated. .github/workflows/publish-sdk.yml runs on every push to main that touches this package, and publishes to PyPI if the version in pyproject.toml is new.

To cut a release:

1. Bump version in pyproject.toml
2. Update CHANGELOG notes in the commit message
3. Merge to main

The workflow runs the SDK suite and the server parity suite before publishing, so a broken version never reaches PyPI. If the version already exists on PyPI the publish is skipped, which means a merge that only edits docs will not fail. Re-running a failed release is possible from the Actions tab via Run workflow.

Publishing uses PyPI Trusted Publishing (OIDC), so no API token is stored in this repository.

License

MIT