Driftstack DRIFTSTACK docs
Docs

Python quickstart

— laser-focused 5-minute path to a working Python Driftstack session. For the multi-language overview see the combined quickstart.

Prerequisites

  • Python 3.10+ (the SDK uses modern type hints + structural matches).
  • Any paid Driftstack tier, including Manual. Free is supported through the desktop app, whose browser sign-in automatically stores a restricted ds_test_… device credential; that credential is not a general SDK or sandbox key.
  • A ds_live_… customer API key. Mint one at app.driftstack.dev/api-keys.

1. Install

pip install driftstack-sdk
# or: uv add driftstack-sdk
# or: poetry add driftstack-sdk

The package ships both sync (Driftstack) and async (AsyncDriftstack) clients off the same wire shape. Pick whichever matches your runtime.

2. Configure the client

Sync (most common, integrates with Flask / Django / scripts):

import os
from driftstack import Driftstack

client = Driftstack(api_key=os.environ["DRIFTSTACK_API_KEY"])
# Optional: base_url="https://staging.driftstack.dev"

Async (FastAPI / Starlette / asyncio scripts):

import asyncio
import os
from driftstack import AsyncDriftstack

async def main():
    async with AsyncDriftstack(api_key=os.environ["DRIFTSTACK_API_KEY"]) as client:
        ...

The async client is httpx.AsyncClient-backed and only opens the connection pool inside async with. The sync client uses a synchronous httpx.Client and supports the context-manager pattern (with Driftstack(...) as client:) for explicit pool cleanup.

3. Run a session

Sync:

import os
from driftstack import Driftstack

with Driftstack(api_key=os.environ["DRIFTSTACK_API_KEY"]) as client:
    session = client.sessions.create({"label": "demo"})
    sid = str(session.id)

    try:
        client.sessions.navigate(sid, {"url": "https://example.com"})
        screenshot = client.sessions.capture(sid, {"kind": "screenshot"})
        # capture() returns a CaptureResponse model — attribute access, not
        # subscript: .kind / .data / .encoding / .byte_size / .duration_ms.
        # For a screenshot, .data is the PNG base64-encoded.
        print(f"captured {screenshot.byte_size} bytes ({screenshot.encoding})")

        state = client.sessions.get_state(sid)
        print("url:", state.url, "title:", state.title)
    finally:
        client.sessions.destroy(sid)

Async:

import asyncio
import os
from driftstack import AsyncDriftstack

async def main():
    async with AsyncDriftstack(api_key=os.environ["DRIFTSTACK_API_KEY"]) as client:
        session = await client.sessions.create({"label": "demo"})
        sid = str(session.id)
        try:
            await client.sessions.navigate(sid, {"url": "https://example.com"})
            screenshot = await client.sessions.capture(sid, {"kind": "screenshot"})
            print(f"captured {screenshot.byte_size} bytes ({screenshot.encoding})")
        finally:
            await client.sessions.destroy(sid)

asyncio.run(main())

4. Error handling

The SDK raises typed exceptions for server errors. Catch DriftstackError and inspect .status (HTTP) or .problem_type (RFC 9457). Other RFC 9457 fields are on the parsed .problem dict:

from driftstack.errors import DriftstackError

try:
    client.sessions.create({"label": "demo"})
except DriftstackError as err:
    if err.status == 429 and (err.problem_type or "").endswith("/tier-limit"):
        # Tier usage quota reached (not the concurrency cap — that is
        # /concurrency-limit). Wait + retry, or upgrade.
        print("cap reached:", err.problem.get("detail"))
    elif err.status == 403 and "apiAccess" in str(err.problem.get("detail", "")):
        # Free keeps desktop access, but customer API/SDK access is paid.
        # Upgrade to resume this key unless it was revoked or expired.
        print(err.problem.get("detail"))
    elif err.status == 401:
        print("bad API key")
    else:
        print("driftstack error:", err.problem_type, err.problem)

For granular handling, catch the subclass directly (RateLimitError, ConcurrencyLimitError, QuotaExceededError, SessionDestroyedError, …). The full mapping lives at /reference/errors.

5. Webhooks (optional)

from driftstack import verify_webhook_signature

ok = verify_webhook_signature(
    body=raw_body,
    header=request.headers["x-driftstack-signature"],
    secret=os.environ["DRIFTSTACK_WEBHOOK_SECRET"],
)
if not ok:
    return Response("invalid signature", status_code=401)

During the 24h signing-secret rotation grace window the single x-driftstack-signature header carries both the new and old HMACs as two v1= entries (t=…,v1=<new>,v1=<old>) — see /webhooks/endpoints for the rotate-secret endpoint. verify_webhook_signature already checks every v1= entry in that header, so the call above keeps verifying through a rotation with no extra arguments while you roll the new secret across your verifier infra.

Pair-mode takeover (interactive AI sessions)

For sessions where a human needs to step in mid-flight:

# Create a pair-mode session, or switch an existing AI session.
session = client.agent_sessions.create({"mode": "pair"})
# OR: client.agent_sessions.set_mode(session_id, "pair")

# When a dashboard user clicks the live preview, the first input-
# event automatically fires the takeover-request transition. Pass
# client_id (any string identifying the calling tab / bot):
result = client.agent_sessions.send_input_event(
    session["id"],
    {"type": "mouseDown", "x": 200, "y": 150, "button": 0},
    client_id="ops-dashboard-tab-a",
)
if result["kind"] == "pair-mode-takeover-fired":
    # result["pair_mode_state"]["kind"] == "takeover-pending"
    pass

# Programmatic takeover from your own ops tooling:
after = client.agent_sessions.takeover(session["id"], "cli-bot")
print(after["pair_mode_state"]["kind"])  # takeover-pending

# Hand control back when done:
back = client.agent_sessions.handback(session["id"])
print(back["pair_mode_state"]["kind"])  # handback-pending

Async mirrors are 1:1: await aclient.agent_sessions.set_mode(...), await aclient.agent_sessions.send_input_event(...), await aclient.agent_sessions.takeover(...), await aclient.agent_sessions.handback(...).

State machine kinds you’ll see: ai-driving, takeover-pending, takeover-queued (mid-decompose deferral), human-driving, handback-pending, handback-queued.

⚠️ The handback half of this loop cannot complete on any deployment today. takeover() works and parks the session in takeover-pending. Advancing from there to human-driving requires the takeover-grant transition, and nothing emits it — the harness has no control-plane surface to fire it yet (tracked in docs/internal/cross-agent-control-plane-contract.md). Consequences you will actually observe:

  • handback() returns 409 pair-mode-conflict every time, because handback-request is only accepted from human-driving.
  • human-driving, handback-pending and handback-queued are unreachable, so a UI that branches on them is dead code.
  • After 30s without a client heartbeat the sweep silently returns the session to ai-driving, so a parked takeover expires on its own.

Drive live sessions through the desktop Simulator’s control channel until the emitter ships.

Modifier vocabulary

keyDown / keyUp events accept a modifiers array. Use the canonical 4-name set — these map 1:1 onto Quartz CGEventFlags on the macOS harness side:

from driftstack.resources.agent_sessions import CANONICAL_MODIFIER_NAMES

result = client.agent_sessions.send_input_event(
    session["id"],
    {"type": "keyDown", "key": "k", "modifiers": ["cmd", "shift"]},
    client_id="dashboard-tab-a",
)

DOM-standard names (Shift / Control / Alt / Meta) round-trip through the schema unchanged but the harness decoder drops them.

Next steps

  • Session lifecycle reference — states, the free-tier 20-minute duration cap, reconnect semantics.
  • Profile management — persistent identity slots that survive across sessions.
  • Agent sessions — natural-language decompose-and-execute on top of the regular driver surface; AI / manual / pair modes, live SSE transcript stream, and the LiveKit-based live video subscription (auto-populated livekit field on session-create, or re-mint via client.agent_sessions.livekit_token(id) after the 24h token TTL — also available on the async client).
  • Bundled LLM and BYOK Anthropic — the two LLM rails agent sessions can use.
  • Idempotency keys — Idempotency-Key is honoured on agent sessions and the billing checkouts, and NOT on POST /v1/sessions; retrying that one mints a second session.
  • Webhook event catalog — every event the platform can push.
  • Error catalogue — every problem-type you might see and how to react.

Stuck? Email support@driftstack.dev with your account id (acc_…) and the failing x-request-id.