-
Notifications
You must be signed in to change notification settings - Fork 0
Initialization
The SDK entry point is Core. You pass an SDKConfig at construction time and
call initialize(). Once core.is_ready is True, you can create visitor
contexts.
Public symbols referenced here are importable from convert_sdk —
Core, SDKConfig, TransportConfig, RefreshConfig.
Use sdk_key when you want the SDK to fetch the project config from the Convert
CDN at startup. The fetch is synchronous and blocking; network failures raise
ConfigLoadError.
import os
from convert_sdk import Core, SDKConfig, TransportConfig
core = Core(
SDKConfig(
sdk_key=os.environ["CONVERT_SDK_KEY"],
environment="production",
transport=TransportConfig(
base_url="https://cdn-4.convertexperiments.com",
timeout=5.0,
),
)
).initialize()
assert core.is_readyThe TransportConfig.auth_secret, if present, is sent as a Bearer token in
the Authorization header when fetching config and delivering tracking events.
Use data when you want to supply the project config yourself — from a file, a
cache, or an inline dict. No network call is made. This mode is ideal for tests,
CI, and local development.
from convert_sdk import Core, SDKConfig
project_config = {
"account_id": "1001",
"project": {"id": "2002", "name": "Demo"},
"features": [],
"experiences": [],
"goals": [],
}
core = Core(SDKConfig(data=project_config)).initialize()
assert core.is_ready| Field | Type | Default | Purpose |
|---|---|---|---|
sdk_key |
str | None |
None |
Convert project SDK key (remote mode) |
data |
dict | None |
None |
Inline config payload (direct/offline mode) |
environment |
str | None |
None |
Environment filter applied to experience eligibility |
cache_level |
str | None |
None |
"low" appends a low-cache hint to the config route |
transport |
TransportConfig |
see below | Network settings for config-fetch and tracking |
batch_size |
int |
10 |
Events per tracking batch before auto-release |
auto_flush_interval_ms |
int | None |
None |
Opt-in periodic flush in milliseconds |
data_store |
DataStore | None |
None |
Optional persistence adapter |
logger |
logging.Logger | None |
None |
Optional caller-supplied logger |
refresh |
RefreshConfig | None |
None |
Opt-in background config refresh |
Exactly one of sdk_key or data must be provided; passing neither raises
InvalidConfigError.
Note: there is no TrackingConfig class — tracking queue settings are
fields directly on SDKConfig (batch_size, auto_flush_interval_ms).
| Field | Default | Purpose |
|---|---|---|
base_url |
https://cdn-4.convertexperiments.com |
HTTPS base URL for config-fetch. Non-HTTPS raises TransportError. |
timeout |
10.0 |
Per-request timeout in seconds |
auth_secret |
None |
Optional bearer token for config and tracking requests |
headers |
{} |
Extra HTTP headers appended to every config request |
verify_tls |
True |
Whether to verify TLS certificates |
Core is a context manager — resources are released cleanly on exit:
from convert_sdk import Core, SDKConfig
with Core(SDKConfig(data=project_config)).initialize() as core:
context = core.create_context("visitor-001")
result = context.run_experience("checkout-experiment")
# Core.close() called automatically on exit| Error | When raised |
|---|---|
InvalidConfigError |
Neither sdk_key nor data was provided, or the payload shape is invalid |
ConfigLoadError |
Network or HTTP error while fetching config with sdk_key
|
TransportError |
Non-HTTPS base_url or other transport configuration failure |
All three are subclasses of ConvertSDKError and carry .code and .context
attributes for structured error handling:
from convert_sdk import ConfigLoadError
try:
core = Core(SDKConfig(sdk_key=os.environ["CONVERT_SDK_KEY"])).initialize()
except ConfigLoadError as exc:
print(exc.code, exc.context)Core is designed to be a long-lived singleton. Create one instance at
application startup and reuse it across requests. The tracking queue owned by
Core is thread-safe; concurrent track_conversion() and core.flush() calls
from different threads will not corrupt it.
# application startup
core = Core(SDKConfig(data=project_config)).initialize()
# per-request handler
def handle_request(visitor_id: str) -> None:
context = core.create_context(visitor_id)
result = context.run_experience("checkout-flow")
...Long-running services can opt into background config refresh by passing a
RefreshConfig to SDKConfig.refresh. Without a RefreshConfig, no
background activity runs and behavior is identical to the MVP — the
default is refresh=None.
import os
from convert_sdk import Core, SDKConfig, RefreshConfig
core = Core(
SDKConfig(
sdk_key=os.environ["CONVERT_SDK_KEY"],
environment="production",
refresh=RefreshConfig(
interval_seconds=300.0, # refresh every 5 minutes
jitter_seconds=30.0, # +/- 30s to avoid herding instances
backoff_factor=2.0, # exponential backoff on failures
backoff_max_seconds=600.0, # cap retries at 10 minutes apart
),
)
).initialize()- Refresh runs on a daemon thread inside
Core. The thread starts atinitialize()time and stops atcore.close()(or when the host process exits). - Each successful refresh that produces a different snapshot replaces the live snapshot through a single atomic swap. In-flight evaluations see either the old or new snapshot, never a partial state.
- Background failures never raise into the host process. The worker thread is daemon-mode, so it does not block process exit.
On every successful swap the SDK emits LifecycleEvent.CONFIG_UPDATED on the
event bus. Subscribe to bust any caches you derive from config:
from convert_sdk import LifecycleEvent
core.on(LifecycleEvent.CONFIG_UPDATED, lambda payload, error=None: my_cache.clear())Call core.refresh_now() to trigger an attempt immediately:
core.refresh_now() # fire-and-forgetWith refresh disabled (SDKConfig.refresh=None), refresh_now() is a no-op.
Refreshes update the snapshot for new contexts. Existing Context objects
retain whatever snapshot was current when they were created — a Context
represents a coherent view of the project for the duration of a request or unit
of work.
- One refresher per
Coreinstance. - Auto-refresh under
os.fork()withoutexecis not supported — the forked child inherits a stopped daemon thread. In pre-fork servers (Gunicorn, Celery prefork) re-initialize the SDK in each worker process after fork. -
Core.close()and the context-manager form stop the refresher cleanly.
If SDKConfig.data is set, there is no remote endpoint to refresh from.
Setting refresh=RefreshConfig(...) in this mode logs a refresh.skipped
diagnostic and does not start a worker.
-
Configuration — full
SDKConfig/TransportConfig/RefreshConfigreference -
Code Examples —
run_experience,run_feature,track_conversion,core.flush - Extending — swap the transport, data store, or observe lifecycle events
Copyrights © 2025 All Rights Reserved by Convert Insights, Inc.
Getting Started
Python SDK
- Quickstart
- Installation
- Initialization
- Configuration
- Code Examples
- Type Hints
- Diagnostics
- Extending
- Testing
- Async & Frameworks
Migration
Core Concepts
- Experiences & Variations
- Feature Flags
- Bucketing Algorithm
- Rule Evaluation
- Segments
- Data Management
- Event System
- API Communication
How-To Guides
- Running Experiences
- Running Features
- Tracking Conversions
- Visitor Context
- Persistent DataStore
- Troubleshooting
Edge & Integrations
Maintainers