Extending the SDK

The SDK exposes exactly three @runtime_checkable typing.Protocols as
swappable I/O seams, plus one optional capability Protocol a transport may
additionally satisfy. Implementing the methods of a protocol is all that is
required — no subclassing, no registration. Each seam has a default adapter the
SDK wires automatically; you replace only what you need to replace.

Seam	Protocol	Default adapter	Injection point
HTTP transport	`convert_sdk.ports.transport.Transport`	`httpx`-backed `HttpxTransport`	`Core(config, transport=...)` (keyword-only)
Visitor state storage	`convert_sdk.ports.storage.DataStore`	`InMemoryDataStore` (per-process)	`SDKConfig(data_store=...)` (config field)
Lifecycle event bus	`convert_sdk.ports.event_bus.EventBus`	In-process synchronous bus	Internal — observe via `Core.on(...)`
(optional) preview fetch	`convert_sdk.ports.transport.SupportsPreviewFetch`	`HttpxTransport` implements it	Same object you pass as `transport=`

There is no logger Protocol. The logging seam is Python's stdlib
logging — supply a logging.Logger via SDKConfig.logger. The SDK emits
through it; it never adds handlers, sets levels, or calls
logging.basicConfig().

Where each seam is injected

The two external seams have deliberately different injection points:

Transport — keyword-only argument on Core:
Core(config, transport=my_transport)
DataStore — field on SDKConfig:
SDKConfig(data=..., data_store=my_store)

The asymmetry is intentional: Core owns the transport lifecycle (it calls
close() and uses it as a context manager); SDKConfig owns the storage
declaration so it can be shared across multiple Core instances in the same
process.

Custom Transport

Implement Transport to replace config fetching and tracking delivery.
Common reasons: mTLS, proxy routing, stubbing in integration tests, or
capturing tracking calls in test assertions.

Protocol requirements

from convert_sdk.ports.transport import Transport
from typing import Any, Dict

class MyTransport:
    def fetch_config(self, config: "SDKConfig") -> Dict[str, Any]:
        """Fetch and return the raw config payload.

        Raise convert_sdk.ConfigLoadError on any transport/HTTP/decode failure.
        """
        ...

    def send_tracking(self, payload: Dict[str, Any], *, sdk_key: str) -> None:
        """Deliver a serialized tracking-events batch over HTTPS.

        sdk_key is keyword-only. Raise a ConvertSDKError subclass on failure.
        """
        ...

    def close(self) -> None:
        """Release any held resources (e.g. HTTP connection pool)."""
        ...

    def __enter__(self) -> "Transport":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

Transport is @runtime_checkable — the __enter__/__exit__ pair is
required for isinstance(obj, Transport) to return True.

Method	Signature	Notes
`fetch_config`	`(config: SDKConfig) -> Dict[str, Any]`	Returns decoded JSON as dict
`send_tracking`	`(payload: Dict[str, Any], *, sdk_key: str) -> None`	`sdk_key` is keyword-only
`close`	`() -> None`	Called on `Core.close()` and at context-manager exit
`__enter__` / `__exit__`	context-manager pair	Required for structural `isinstance` check

Complete example: in-memory stub transport

from typing import Any, Dict
from convert_sdk import Core, SDKConfig
from convert_sdk.ports.transport import Transport

class StubTransport:
    """Returns canned config and captures tracking payloads."""

    def __init__(self, config_payload: Dict[str, Any]) -> None:
        self._config = config_payload
        self.tracking_calls: list = []

    def fetch_config(self, config) -> Dict[str, Any]:
        return self._config

    def send_tracking(self, payload: Dict[str, Any], *, sdk_key: str) -> None:
        self.tracking_calls.append(payload)

    def close(self) -> None:
        pass

    def __enter__(self) -> "StubTransport":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

stub = StubTransport(config_payload={
    "account_id": "1001",
    "project": {"id": "2002", "name": "Test"},
    "experiences": [],
    "features": [],
    "goals": [],
})

# @runtime_checkable — duck-typed conformance check:
assert isinstance(stub, Transport)

# transport= is KEYWORD-ONLY on Core:
core = Core(SDKConfig(sdk_key="test-key"), transport=stub).initialize()
context = core.create_context("visitor-001")
context.track_conversion("purchase_completed")
core.flush()

assert stub.tracking_calls   # our stub handled the delivery
core.close()

Example: retrying transport wrapper

import time
from typing import Any, Dict
from convert_sdk.ports.transport import Transport
from convert_sdk.errors import ConfigLoadError

class RetryingTransport:
    """Wraps another transport and retries fetch_config up to `max_retries` times."""

    def __init__(self, inner: Transport, max_retries: int = 3) -> None:
        self._inner = inner
        self._max_retries = max_retries

    def fetch_config(self, config) -> Dict[str, Any]:
        last_exc: Exception | None = None
        for attempt in range(self._max_retries):
            try:
                return self._inner.fetch_config(config)
            except ConfigLoadError as exc:
                last_exc = exc
                if attempt < self._max_retries - 1:
                    time.sleep(2 ** attempt)
        raise last_exc  # type: ignore[misc]

    def send_tracking(self, payload: Dict[str, Any], *, sdk_key: str) -> None:
        self._inner.send_tracking(payload, sdk_key=sdk_key)

    def close(self) -> None:
        self._inner.close()

    def __enter__(self) -> "RetryingTransport":
        return self

    def __exit__(self, *exc: Any) -> None:
        self.close()

Optional: supporting preview fetch

Context.set_preview() can force a variation for an experiment that is not
in the loaded config — a draft or paused one reached through a preview link. To
resolve it, the SDK asks the transport for that single experience via
?exp={experience_id}.

That capability lives in its own Protocol rather than on Transport, because
adding a required method to a public extension point would break every existing
custom transport. Implement the one extra method and your transport opts in
automatically through structural typing:

Taking the StubTransport from above and adding one method to it:

from typing import Any, Dict
from convert_sdk.ports.transport import Transport, SupportsPreviewFetch

class PreviewCapableStubTransport(StubTransport):
    def fetch_config_by_experience(
        self, config, experience_id: str
    ) -> Dict[str, Any]:
        """Return the decoded config body scoped to one experience.

        Raise convert_sdk.ConfigLoadError on any transport/HTTP/decode failure —
        Context.set_preview catches it and degrades to inert-with-warning.
        """
        return self._config

transport = PreviewCapableStubTransport(config_payload={...})
assert isinstance(transport, Transport)             # the required seam
assert isinstance(transport, SupportsPreviewFetch)  # the optional capability

Subclassing StubTransport is just brevity for the example — neither Protocol
requires a base class. Any object carrying the methods satisfies both.

Core duck-checks isinstance(transport, SupportsPreviewFetch) before
attempting the fetch. Not implementing it is a supported outcome, not an
error — the fetch is simply never attempted, and a preview target absent from
the loaded snapshot degrades to set_preview's inert-with-warning contract. The
same degradation applies in direct-config (data=) mode, where there is no
remote endpoint to query at all.

The bundled HttpxTransport already implements the method. It memoizes each
resolved (sdk_key, experience_id) body process-wide for 60 seconds; a failed
fetch is never memoized, so a transient error cannot strand a QA session for the
whole window. If you write your own, caching is your call — the SDK adds none
around the Protocol.

Custom DataStore

Implement DataStore to persist visitor state and deduplication markers across
process restarts — for example, backed by Redis, Memcache, or a database. Inject
it through SDKConfig.data_store, not as a Core argument.

Protocol requirements

from convert_sdk import DataStore   # top-level export
from typing import Any, Optional

class MyStore:
    def get(self, key: str) -> Any:
        """Return the value stored under key, or None if absent or expired."""
        ...

    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        """Store value. ttl is expiry in seconds; None means no expiry."""
        ...

    def has(self, key: str) -> bool:
        """True if key has a present, unexpired value."""
        ...

    def delete(self, key: str) -> None:
        """Remove key. Deleting an absent key is a safe no-op."""
        ...

DataStore is @runtime_checkable — isinstance(obj, DataStore) checks for
the four method names at runtime (not their signatures).

Method	Signature	Notes
`get`	`(key: str) -> Any`	Returns `None` for absent or expired keys
`set`	`(key: str, value: Any, ttl: Optional[int] = None) -> None`	`ttl` in seconds; `None` = no expiry
`has`	`(key: str) -> bool`	Must return `False` for absent and expired keys
`delete`	`(key: str) -> None`	Idempotent — no-op on absent key

Implementations must be safe to call from the synchronous tracking path and
from worker threads (the SDK's flush timer runs on a daemon thread).

Complete example: dict-backed store

from typing import Any, Optional
from convert_sdk import Core, SDKConfig, DataStore

class DictStore:
    def __init__(self) -> None:
        self._data: dict = {}
        self.writes: int = 0

    def get(self, key: str) -> Any:
        return self._data.get(key)

    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        self._data[key] = value
        self.writes += 1

    def has(self, key: str) -> bool:
        return key in self._data

    def delete(self, key: str) -> None:
        self._data.pop(key, None)

store = DictStore()
assert isinstance(store, DataStore)

# data_store is a CONFIG FIELD, not a Core argument:
core = Core(SDKConfig(data=my_config, data_store=store)).initialize()
context = core.create_context("visitor-001")
context.set_segments({"loyalty_tier": "gold"})  # persists through the store

assert store.writes >= 1
core.close()

Example: Redis-backed store

import json
from typing import Any, Optional
from convert_sdk import Core, SDKConfig

class RedisDataStore:
    def __init__(self, redis_client) -> None:
        self._redis = redis_client

    def get(self, key: str) -> Any:
        raw = self._redis.get(key)
        return json.loads(raw) if raw is not None else None

    def set(self, key: str, value: Any, ttl: Optional[int] = None) -> None:
        serialized = json.dumps(value)
        if ttl is not None:
            self._redis.setex(key, ttl, serialized)
        else:
            self._redis.set(key, serialized)

    def has(self, key: str) -> bool:
        return bool(self._redis.exists(key))

    def delete(self, key: str) -> None:
        self._redis.delete(key)

core = Core(
    SDKConfig(data=my_config, data_store=RedisDataStore(redis_client=my_redis))
).initialize()

The built-in InMemoryDataStore is thread-safe but process-local. With the
default store, deduplication markers reset on process restart. A shared Redis
store removes that isolation without any protocol change.

Namespaced storage keys

The SDK stores different types of state under distinct key prefixes so they
never collide in the shared store:

Visitor state: keys built by
convert_sdk.ports.storage.visitor_state_key(visitor_id) →
"state:[\"visitor-001\"]"
Deduplication markers: keys built by the tracking layer, prefixed "dedup:"

The visitor-state value is a structured envelope with three independent maps —
{"attributes": {...}, "segments": {...}, "bucketing": {...}} — so a later
create_context(visitor_id) rehydrates stored attributes, default segments, and
the sticky-bucketing decision map ({experience_id: variation_id}) through the
same key. A store that round-trips JSON-serializable dicts needs no awareness of
the envelope's shape.

from convert_sdk.ports.storage import visitor_state_key

key = visitor_state_key("visitor-001")
# key == 'state:["visitor-001"]'

See Persistent DataStore for the full lifecycle of
reads and writes.

Logging seam (no Protocol)

There is no Logger Protocol — the seam is Python's stdlib logging. Pass
your own logging.Logger via SDKConfig.logger to route SDK output through
your application's infrastructure:

import logging
from convert_sdk import Core, SDKConfig

my_logger = logging.getLogger("myapp.convert")
core = Core(SDKConfig(data=my_config, logger=my_logger)).initialize()

To silence all SDK output in production:

logging.getLogger("convert_sdk").setLevel(logging.WARNING)

The SDK never adds handlers, sets levels, or calls logging.basicConfig() —
that is always the application's responsibility.

Observing lifecycle events (EventBus)

You do not implement the EventBus Protocol directly. The SDK owns the bus.
You observe it by registering handlers through Core.on:

from convert_sdk import LifecycleEvent
from convert_sdk.events import QueueReleasedPayload, ConversionEventPayload

def on_conversion(payload: ConversionEventPayload, error=None) -> None:
    print(f"conversion: goal={payload.goal_key} visitor={payload.visitor_id}")

def on_released(payload: QueueReleasedPayload, error=None) -> None:
    if error is not None:
        print(f"delivery failed: status={payload.status_code}")
    else:
        print(f"released {payload.batch_size} events, reason={payload.reason.value}")

core.on(LifecycleEvent.CONVERSION, on_conversion)
core.on(LifecycleEvent.API_QUEUE_RELEASED, on_released)

Handlers receive (payload, error=None) — error is a BaseException | None.
A handler that raises is isolated, logged, and swallowed — it cannot break
delivery or other handlers.

EventBus Protocol (for custom bus implementations)

If you need to replace the bus (advanced — most integrations only need
Core.on):

from convert_sdk.ports.event_bus import EventBus, EventHandler
from convert_sdk.events import LifecycleEvent
from typing import Any, Optional

class MyEventBus:
    def on(self, event: LifecycleEvent, handler: EventHandler) -> None:
        ...

    def emit(
        self,
        event: LifecycleEvent,
        payload: Any,
        error: Optional[BaseException] = None,
    ) -> None:
        ...

Note: EventBus has exactly two methods — on and emit. There is no
off (handler deregistration is not part of the MVP protocol surface).

See Event System for the full lifecycle event reference.

Testing with custom adapters

The most common use of the extension points in tests: a stub transport with
canned config payloads avoids network calls; a fresh InMemoryDataStore per
test isolates deduplication state.

from typing import Any, Dict
from convert_sdk import Core, SDKConfig, InMemoryDataStore

class CannedTransport:
    def __init__(self, payload: Dict[str, Any]) -> None:
        self._payload = payload
        self.tracking_calls: list = []

    def fetch_config(self, config) -> Dict[str, Any]:
        return self._payload

    def send_tracking(self, payload: Dict[str, Any], *, sdk_key: str) -> None:
        self.tracking_calls.append(payload)

    def close(self) -> None:
        pass

    def __enter__(self) -> "CannedTransport":
        return self

    def __exit__(self, *exc: Any) -> None:
        pass

def make_test_core(config_payload: Dict[str, Any]) -> Core:
    return Core(
        SDKConfig(sdk_key="test-key", data_store=InMemoryDataStore()),
        transport=CannedTransport(config_payload),
    ).initialize()

Future async support

These Protocol seams are the foundation for the planned Phase 3 async surface.
When async lands, each seam gains an async sibling (e.g. AsyncTransport,
AsyncDataStore) rather than being re-shaped — existing sync adapters keep
working unchanged. See the async design intent (not implemented in the current release).

Where each seam is injected

Custom Transport

Protocol requirements

Complete example: in-memory stub transport

Example: retrying transport wrapper

Optional: supporting preview fetch

Custom DataStore

Protocol requirements

Complete example: dict-backed store

Example: Redis-backed store

Namespaced storage keys

Logging seam (no Protocol)

Observing lifecycle events (EventBus)

EventBus Protocol (for custom bus implementations)

Testing with custom adapters

Future async support

What to read next