Python SDK quickstart

The driftstack Python package wraps the same REST surface as the curl quickstart. It supports both synchronous and asyncio call-styles from a single client, so it slots into Django, Flask, FastAPI, or a plain script equally well.

1. Install

pip install driftstack-sdk
# or, with poetry / uv:
poetry add driftstack-sdk
uv add driftstack-sdk

Python ≥ 3.10 is supported. The package ships type stubs; mypy --strict works against the public surface without further config.

2. Construct the client

import os
from driftstack import Driftstack

client = Driftstack(api_key=os.environ["DRIFTSTACK_API_KEY"])
# base_url kwarg overrides the default https://api.driftstack.dev
# for staging or self-hosted deployments.

Reuse one client across your process. It is internally pooled via httpx and safe for concurrent use across threads or asyncio tasks.

3. Start a session

session = client.sessions.create({
    "archetype": "default",
    "purpose": "production_customer",
    "label": "demo",
})
print(session["id"], session["status"], session["archetype"])

The session-create call does not take a target URL. Drive the session to a URL with client.sessions.navigate(session_id, {"url": "https://example.com"}). Session ids are prefixed ses_.

4. Drive and finish the session

There is no built-in “wait until terminal” helper — drive the session with the methods that fit your workflow (navigate, interact, wait, capture), then destroy when done. The wait method blocks on a server-side condition (selector, URL, time):

client.sessions.navigate(session["id"], {"url": "https://example.com"})
client.sessions.wait(session["id"], {"condition": {"kind": "time", "ms": 1000}})
shot = client.sessions.capture(session["id"], {"kind": "screenshot"})
print("captured", shot["byte_size"], "bytes")
client.sessions.destroy(session["id"])  # idempotent

For batch workloads where you don't need to drive the session moment-to-moment, prefer webhooks over polling.

5. Paginate over historical sessions

# Sync — lazy iterator over cursor pages.
for s in client.sessions.iterate(limit=50):
    if s["status"] == "destroyed":
        print(s["id"])

# Async — same shape with AsyncDriftstack.
import asyncio
from driftstack import AsyncDriftstack

async def main() -> None:
    async with AsyncDriftstack(api_key=os.environ["DRIFTSTACK_API_KEY"]) as ac:
        async for s in ac.sessions.iterate(limit=50):
            print(s["id"])

asyncio.run(main())

iterate walks every cursor page and stops on next_cursor: null. The single-page form is client.sessions.list({"limit": ..., "cursor": ...}).

Error handling

Every SDK method raises a DriftstackError subclass on non-2xx responses. The exception carries status, problem_type (the RFC 7807 URI), message, and problem (the full parsed problem dict, e.g. exc.problem["retry_after_seconds"]). Branch on the subclass for clean dispatch:

from driftstack import DriftstackError
from driftstack.errors import RateLimitError, ValidationError

try:
    client.sessions.create({"archetype": "definitely-not-a-real-archetype"})
except ValidationError as exc:
    # per-field issues live on exc.problem
    ...
except RateLimitError as exc:
    # exc.problem["retry_after_seconds"] is set
    ...
    else:
        raise

Sync ↔ async parity

Every method on Driftstack has a mirror on AsyncDriftstack. The async variant returns coroutines and async iterators; the sync variant blocks the caller. Pick the one that fits your runtime; do not mix them in a single call-chain.

Where to go next

• Full API reference — every endpoint, every field.
• Crypto orders — customer-facing crypto-checkout surface.
• TypeScript SDK — same surface, TS-native.
• Webhooks — avoid polling in production.
• Cost monitoring — set alerts before the bill surprises you.
• Error codes — the stable RFC 7807 type URIs the SDK's exception subclasses carry.

Need help?

[email protected]. We respond within one business day.