Skip to main content

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.

SDK automation needs a ds_live_… customer API key. Every plan includes the API, the free trial too.

1. Install

python -m pip install driftstack-sdk

The Python SDK is published on PyPI and remains pre-1.0 with an Alpha classifier. Pin a compatible version in your requirements or lockfile for reproducible deployments. 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. Run your first agent session

An agent session is an iPhone Safari browser in the cloud that an AI drives for you: your code sends 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). You will need: 

import time
import uuid

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"])

agent_sessions.create() started an agent session through the proxy you named; without a proxy_id it is refused with 422 proxy-required, and nothing is created. A new session reads status: active at once; the loop waits for ready: true, which comes when its iPhone Safari browser has started. message() gave the agent the task: kind is plan-executed when it ran, with the answer it read back, and the idempotency key makes a retried message the same turn rather than a second one. close() ended the session and released its slot. Agent session ids are prefixed agt_.

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. Paginate over your agent sessions

# Sync — lazy iterator over cursor pages, newest first.
for s in client.agent_sessions.iterate(limit=50):
    if s["status"] == "closed":
        print(s["id"], s["closed_reason"])

# 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.agent_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.agent_sessions.list(limit=..., cursor=...). From 0.6.0 (not published yet), both take status (for example status="open") to list only the sessions with that status; iterate keeps it on every page.

Error handling

Every SDK method raises a DriftstackError subclass on non-2xx responses. The exception carries status, problem_type (the RFC 9457 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.errors import ProxyRequiredError, RateLimitError

try:
    client.agent_sessions.create({"mode": "ai", "proxy_id": os.environ["DRIFTSTACK_PROXY_ID"]})
except ProxyRequiredError as exc:
    # no proxy_id, so nothing was created
    ...
except RateLimitError as exc:
    # exc.problem["retry_after_seconds"] is set
    ...

Driving each step yourself

client.sessions is the direct session API: your code drives each step itself with navigate, interact, wait and capture instead of giving an AI a task. On hosted Driftstack only agent sessions load pages, so use client.agent_sessions for real browsing. See the Sessions reference.

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

Need help?

[email protected]. Reply times follow your plan's support on the pricing page: a 48-hour reply target on Starter and Pro.