Drıftstack docs
Docs

Python quickstart

A five-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).
  • driftstack-sdk 0.5.0 or newer. The examples on this page are written against it.
  • A paid plan: these steps use the AI agent, which the free trial does not include.
  • A ds_live_… customer API key. Mint one at app.driftstack.io/api-keys.
  • The id of one of your saved proxies: every agent session goes out through a proxy you chose. See Find a proxy’s id.
  • A way to pay for the AI: your plan’s monthly AI credits, or your own Anthropic key. See Who pays for the AI.

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 with the same methods. Pick whichever fits your application.

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://api.driftstack.dev" (the default)

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 your first session

An agent session is an iPhone Safari browser in the cloud that an AI drives for you: you send a task in plain words, and the agent plans the taps and page loads, runs them, and answers. Every agent session goes out through one of your saved proxies, so the create names its id (proxy_id).

import os
import time
import uuid
from driftstack import Driftstack

with Driftstack(api_key=os.environ["DRIFTSTACK_API_KEY"]) as client:
    agent = client.agent_sessions.create(
        {"mode": "ai", "proxy_id": os.environ["DRIFTSTACK_PROXY_ID"]}
    )
    try:
        # Wait up to two minutes for its browser: `status` is "active" from the
        # start, so read `ready`.
        for _ in range(60):
            s = client.agent_sessions.get(agent["id"])
            if s["status"] == "closed":
                raise RuntimeError(f"The session did not start: {s['closed_reason']}")
            if s.get("ready") is not False:
                break
            time.sleep(2)
        reply = client.agent_sessions.message(
            agent["id"],
            "Open https://example.com and tell me the page title.",
            idempotency_key=str(uuid.uuid4()),
        )
        if reply["kind"] == "plan-executed":
            print(reply.get("answer", reply["results"]))
        else:
            print(reply["kind"], reply)
    finally:
        # Always close it: an open session counts against your plan's
        # concurrent-session limit.
        client.agent_sessions.close(agent["id"])

The async client (AsyncDriftstack) has the same methods; await each call.

The guide Run AI tasks from your code covers the rest a program needs: answering the agent’s questions, approving a payment it held, stopping a task that runs too long, safe retries, and limits.

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:

import os
from driftstack.errors import DriftstackError, ProxyRequiredError

try:
    client.agent_sessions.create({"mode": "ai", "proxy_id": os.environ["DRIFTSTACK_PROXY_ID"]})
except ProxyRequiredError as err:
    # No proxy_id (or an empty one): nothing was created. Pass the id of one
    # of your saved proxies.
    print(err)
except DriftstackError as err:
    if err.status == 429 and (err.problem_type or "").endswith("/concurrency-limit"):
        # As many sessions open as your plan allows. Close one you created, or wait.
        print("concurrent cap reached:", err.problem.get("detail"))
    elif err.status == 402:
        # The free trial has ended, or a subscription lapsed: the account keeps
        # its data, and sessions start again once it chooses a plan.
        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.

Next steps

  • Run AI tasks from your code — give the AI agent a task, read its answer, and handle questions, approvals, stops, errors and retries.
  • Session lifecycle reference — states, reconnect semantics.
  • Profile management — persistent profiles that survive across sessions.
  • Agent sessions — let an AI agent drive a session from plain-language instructions, with a live transcript stream and live video (the livekit field on the create response).
  • Who pays for the AI — your plan’s monthly AI credits, or your own Anthropic key.
  • Idempotency keys — Idempotency-Key is honoured on agent sessions, 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 [email protected] with your account id (acc_…, from client.account.whoami()) and the failing x-request-id.