Metadata-Version: 2.5
Name: guidinghand
Version: 0.8.3
Summary: The official Python SDK for the GuidingHand API: run tasks on a person's computer with an AI hand.
Project-URL: Homepage, https://guidinghand.ai
Project-URL: API reference, https://guidinghand.ai/openapi.json
Project-URL: Developers, https://guidinghand.ai/#api
Author-email: GuidingHand <dev@guidinghand.ai>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,api,computer-use,guidinghand,sdk
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx>=0.25
Description-Content-Type: text/markdown

# GuidingHand Python SDK

The official Python client for the [GuidingHand](https://guidinghand.ai) API. You create a session and send its invite link to the person at the computer. Once their computer connects, you run tasks on it, answer the agent's questions and approvals, and read back the result and the recording.

```bash
pip install guidinghand
```

Python 3.9 or later. The only dependency is [httpx](https://www.python-httpx.org/).

## Quickstart

Create an org API key in the console under **Settings → API keys**. It starts with `gh_live_` and is shown once.

```bash
export GUIDINGHAND_API_KEY="gh_live_..."
```

```python
from guidinghand import GuidingHand

client = GuidingHand()  # reads GUIDINGHAND_API_KEY

# 1. A session, and the invite link for the customer
session = client.sessions.create(metadata={"ticket_id": "T-1042"})
print("Send this link to the customer:", session["invite_url"])

# 2. Wait for their computer to connect (polls every 3 s, up to 10 minutes by default)
session = client.sessions.wait_for_connection(session["session_id"])
print("Connected:", session["device"]["name"] or session["device"]["os"])

# 3. Run a task, answering its questions and approvals as they come up
#    (leave on_question out to let the customer answer the questions on their screen)
task = client.tasks.run(
    session["session_id"],
    "Turn on Dark Mode",
    on_event=lambda e: print(f"[{e['type']}] {e['message']}"),
    on_question=lambda q: input(f"{q['question']} {q['options']} > "),
    on_approval=lambda a: input(f"Approve {a['action']!r} (risk: {a['risk']})? [y/N] > ").strip().lower() == "y",
)

# 4. The result and the replay
print(f"{task['status']}: {task['result'] or task['error'] or ''}")
print("Replay:", task["replay_url"])
```

Every method returns the API's JSON as plain dicts. They are typed with `TypedDict`s (`Agent`, `Session`, `Task`, `Event`, `Recording`, `Webhook`, `Page`, ...), so editors and type checkers know the keys. Field names are the API's own (`session_id`, `task_id`, `agent_id`, ...), and so are the parameter names.

## The client

```python
from guidinghand import GuidingHand

client = GuidingHand(
    api_key=None,                        # default: the GUIDINGHAND_API_KEY environment variable
    base_url="https://guidinghand.ai",   # or "https://dev.guidinghand.ai" (Stripe test mode)
    timeout=60.0,                        # seconds per request; long polls get their wait on top
    max_retries=2,
)
```

Without an API key, the constructor raises `AuthenticationError`. The client holds a connection pool: use it as a context manager or call `client.close()` when you are done.

```python
with GuidingHand() as client:
    ...
```

`AsyncGuidingHand` has the same resources and methods, and you `await` them. See [Async](#async).

## Agents

An agent holds your instructions, the AI model it runs on and its reasoning effort, the greeting shown on the invite page, and what the person at the computer sees and may do. Every org has a `default` agent that works without setup.

```python
client.agents.list()                     # {"data": [Agent, ...], "has_more", "next_cursor"}: default first, then by name
client.agents.list(q="billing")          # q searches names, agent ids and display names
agent = client.agents.create(
    name="Billing help",
    agent_id="billing",                  # optional: made from the name when left out
    instructions="Our billing app is Acme Billing. Open it from the Dock first.",
    model="claude-sonnet-5-5",           # the AI model; None: GuidingHand's default ("gpt-6-astra")
    effort="medium",                     # "low" | "medium" | "high"
    greeting="Hi, this is Acme support.",
    display_name="Acme Support",         # what the app calls the agent on their screen (default: GuidingHand)
    narration=True,                      # show its thoughts and steps on their screen (default True)
    customer_answers=True,               # they can answer its questions in the app (default True)
    customer_approvals=True,             # they can approve or deny its approval requests in the app (default True)
    guardrails={"block": ["purchase"]},  # rules GuidingHand enforces on every step (below); the rest keep their defaults
)
client.agents.retrieve("billing")
client.agents.update("billing", effort="high")   # only the fields you pass; effort=None resets it
client.agents.update("billing", model="gpt-6.1-sol")   # model=None goes back to the default model
client.agents.update("billing", customer_answers=False)   # only your team answers questions
client.agents.update("billing", customer_approvals=False) # only your team decides approvals
client.agents.delete("billing")          # its sessions switch to default; deleting "default" resets it
```

`model` and `effort` set the agent's price per agent minute; each model's prices are in [Model](https://docs.guidinghand.ai/concepts/agents#model). Tasks carry the `model` and `effort` they ran at, and `cost_cents` once they end. `agent["invite_url_template"]` is the agent's invite link with a `{code}` placeholder. Creating an `agent_id` that already exists raises `ConflictError`.

`guardrails` are rules GuidingHand enforces on every step the agent takes, whatever the model says ([Guardrails](https://docs.guidinghand.ai/concepts/agents#guardrails)). Pass a partial dict: only the keys you give change, a key set to `None` goes back to its default, and `guardrails=None` resets them all. The agent always has the whole object, with defaults filled in.

```python
agent = client.agents.update("billing", guardrails={
    "block": ["purchase", "install"],               # never, even with an approval
    "apps": {"allow": ["Acme Billing", "Safari"]},
    "sites": {"allow": ["acme.com"]},               # and its subdomains; site rules need a Mac for now
    "scope": "Billing help in Acme Billing and on acme.com.",
    "approval_timeout_minutes": 10,                 # then denied, with answered_by "timeout"
})
agent["guardrails"]["max_steps"]   # 150: the defaults are filled in
```

A refused step is a `blocked` event: `message` says what didn't happen and why, and `data["guardrail"]` (a `GuardrailDecision`) has the `rule` and the `decision`. The agent plans around it, and a `stop` decision ends the task.

### Versions

Save changes to an agent's draft without changing what runs, try the draft on a task, then publish it as the next [version](https://docs.guidinghand.ai/concepts/agents#versions):

```python
client.agents.save_draft("billing", instructions="Open Acme Billing, then Settings > Billing.")   # only these fields change
task = client.tasks.run(session_id, "Update the billing address", agent_version="draft")
task["agent_draft"]       # True
agent = client.agents.publish("billing", note="Start from Settings > Billing")
agent["version"]          # the new version: new tasks run it

client.agents.retrieve_draft("billing")             # the agent as its draft would make it (has_draft False: there's none)
client.agents.discard_draft("billing")
client.agents.versions.list("billing")["data"]      # newest first: version, note, published_at, published_by
client.agents.versions.retrieve("billing", 2)["agent"]   # the agent as it was then
client.agents.versions.restore("billing", 2)        # into the draft; publish to make it live again
```

`agents.update` still changes the agent right away, as a new version. `agent_version` also takes a number, to run an earlier version. Publishing with no draft raises `ConflictError`. Each task says which version it ran (`agent_version`, `agent_draft`) and who started it (`created_by`).

### Tools and files

An agent has its own [file system](https://docs.guidinghand.ai/concepts/agents#the-file-system) (on by default), and can call your [HTTP tools](https://docs.guidinghand.ai/concepts/agents#http-tools):

```python
client.agents.update("billing", tools={
    "files": {"access": "read_write"},   # or "read_only"; {"enabled": False} turns it off
    "http": [{
        "name": "lookup_order",
        "description": "Look up an order by its number.",
        "url": "https://api.acme.com/guidinghand/lookup-order",
        "parameters": {"type": "object", "properties": {"order": {"type": "string"}}, "required": ["order"]},
    }],
})

client.agents.files.write("billing", "/guides/returns.md", "# Returns\nWithin 30 days.")
page = client.agents.files.list("billing", folder="/guides")   # by path: {"data": [...], "usage": {"files", "bytes", "limits"}}
client.agents.files.list("billing", q="returns")               # q searches paths
client.agents.files.retrieve("billing", "/guides/returns.md")["content"]
client.agents.files.delete("billing", "/guides/returns.md")
```

GuidingHand signs each tool call with the org's tool secret (`client.tool_secret.retrieve()["secret"]`, and `client.tool_secret.rotate()` for a new one). Check it on your endpoint with the raw body:

```python
from guidinghand import verify_tool_call, WebhookVerificationError

# Flask
@app.post("/guidinghand/lookup-order")
def lookup_order():
    try:
        call = verify_tool_call(request.get_data(), request.headers.get("GuidingHand-Signature"), TOOL_SECRET)
    except WebhookVerificationError:
        return "", 401
    return jsonify(find_order(str(call["arguments"]["order"])))   # the model gets this body
```

`verify_tool_call` works like [`verify_webhook`](#webhooks) and returns the call: `{"tool", "arguments", "task": {"task_id", "session_id", "agent_id", "agent_version"}, "org_id"}`. Each call is a `tool` event in the task.

## Sessions

A session is one pairing code (`session_id`, e.g. `K7QM-24XP`) for one computer. Codes expire after 72 hours without use.

```python
session = client.sessions.create(agent_id="billing", metadata={"ticket_id": "T-1042"})
session["invite_url"]        # send this to the person at the computer
session["status"]            # "waiting" | "connected" | "disconnected" | "expired"

client.sessions.retrieve(session["session_id"])
client.sessions.list(agent_id="billing", limit=20)            # one page, newest first
client.sessions.list(q="T-1042")                              # q searches codes, computer names and metadata
for s in client.sessions.list_all(agent_id="billing"):        # every page
    print(s["session_id"], s["status"], s.get("task_count"))
client.sessions.disconnect(session["session_id"])  # lets the computer go; the code can be entered again
client.sessions.expire(session["session_id"])      # the code, invite link and token stop working
client.sessions.delete(session["session_id"])  # disconnects, stops a running task, deletes its history

session = client.sessions.wait_for_connection(session["session_id"], timeout=600, poll_interval=3)
```

`wait_for_connection` returns the session once `status` is `connected` and `device` is set. It raises `SessionExpired` if the code expires first, and `GuidingHandTimeout` after `timeout` seconds (`timeout=None` waits indefinitely). As an alternative to polling, a [webhook](#webhooks) sends `session.connected`.

## Tasks

A task is a prompt carried out on the session's computer. Each computer runs one task at a time.

### Run a task to the end

`tasks.run` starts a task and follows it until it is done, then returns the finished task.

```python
task = client.tasks.run(
    session_id,
    "Update the billing email to ops@acme.com",
    agent_id=None,          # run a different agent than the session's
    agent_version=None,     # "draft" to try the agent's draft, or an earlier version's number
    metadata={"ticket_id": "T-1042"},
    request_id="T-1042-email",  # idempotency key: the same request_id returns the same task
    on_event=print,         # every event, as it happens
    on_question=lambda q: "Work",   # q = {"type": "question", "question_id", "question", "options", "customer_can_answer"}
    on_approval=lambda a: True,     # a = {"type": "approval", "approval_id", "action", "risk", "customer_can_approve"}
    timeout=None,           # seconds; then the task is stopped and GuidingHandTimeout raised (unless it just finished)
)
task["status"]   # "completed" | "failed" | "stopped"
task["result"]   # the agent's summary when it completed
task["error"]    # why it failed
```

`on_question` returns the answer as a string (or `None` to leave it to the customer, below). `on_approval` returns `True` or `False`, `"approve"` or `"deny"`, or `{"decision": "deny", "note": "Not on a Friday"}` to pass a note to the agent (or `None` to leave it to the customer). With `GuidingHand` the handlers are plain functions: an `async def` handler raises `TypeError`, so use [`AsyncGuidingHand`](#async) for those. If the agent needs a handler you didn't pass, `run` raises `NeedsInput` instead of hanging. The task keeps waiting on the server, so you can answer it yourself:

```python
from guidinghand import NeedsInput

try:
    task = client.tasks.run(session_id, "Turn on Dark Mode")
except NeedsInput as e:
    print(e.pending)   # the question or approval
    client.tasks.respond(e.task["task_id"], question_id=e.pending["question_id"], answer="Work")
```

### Questions and approvals the customer answers

Unless the agent turns it off (`customer_answers=False`), the agent's questions also appear on the customer's screen, and they can answer them there. `pending["customer_can_answer"]` says whether they can answer the open one. The first answer is used, from them or from you.

- Without `on_question`, `run` leaves those questions to the customer and keeps following the task (pass `timeout` to give up eventually). It raises `NeedsInput` only for a question they can't answer.
- With `on_question`, return `None` to leave one to them. If they answer while your handler works, your answer is refused with a 409 and `run` carries on.
- `tasks.respond` after they answered raises `ConflictError` with `e.extra["code"] == "already_answered"` and `e.extra["answered_by"] == "customer"`.
- The `answer` event has `data["answered_by"]` (`"customer"` or `"operator"`), and the `task.question_answered` webhook carries the answer and who gave it.

Approvals work the same way. Unless the agent turns it off (`customer_approvals=False`), the customer can approve or deny the agent's approval requests on their screen, and `pending["customer_can_approve"]` says whether they can decide the open one.

- Without `on_approval`, `run` leaves those approvals to the customer and keeps following. With `on_approval`, return `None` to leave one to them.
- `tasks.respond` after they decided raises `ConflictError` with `e.extra["code"] == "already_answered"` and `e.extra["answered_by"] == "customer"`.
- The `approved` and `denied` events have `data["answered_by"]`, and the `task.approval_decided` webhook carries the decision and who made it. `answered_by` is `"timeout"` when nobody decided within the agent's `approval_timeout_minutes`, so it was denied.

```python
def on_event(e):
    if e["type"] == "answer":
        print(f"{e['data']['answered_by']} answered: {e['message']}")

task = client.tasks.run(
    session_id,
    "Set up the office printer",
    on_event=on_event,
    on_question=lambda q: None if q["customer_can_answer"] else ask_my_team(q["question"], q["options"]),
    on_approval=lambda a: None if a["customer_can_approve"] and a["risk"] != "high" else ask_my_team_to_approve(a["action"]),
)
```

### Follow a task yourself

`tasks.events` long-polls `GET /v1/tasks/{id}/events`. Each page has the new events and the task as it is now, so you can answer what it is waiting on:

```python
from guidinghand import ConflictError

task = client.tasks.create(session_id, "Turn on Dark Mode", request_id="T-1042-dark-mode")
task_id, after = task["task_id"], 0

while True:
    page = client.tasks.events(task_id, after=after, wait_ms=25000)   # waits up to wait_ms (max 55000) for news
    for event in page["data"]:
        print(event["cursor"], event["type"], event["message"])
    after = page["cursor"]           # pass as `after` next time
    task = page["task"]              # the task now
    if task["done"]:
        break
    pending = task["pending"]        # what it is waiting on now, if anything
    try:
        if pending and pending["type"] == "question":
            client.tasks.respond(task_id, question_id=pending["question_id"], answer="Work")
        elif pending and pending["type"] == "approval":
            client.tasks.respond(task_id, approval_id=pending["approval_id"], decision="approve")
    except ConflictError:
        pass                         # already answered (by the customer on their screen, an earlier run, or someone else)

print(task["status"], task["result"])
```

Answer `task["pending"]`, not the `question` and `approval_required` events: the events start from the beginning, so a re-run (the same `request_id` returns the same task) sees questions that were already answered.

To only watch, `stream(task_id, after=0)` yields each event once and ends when the task is done. To resume from a known point, pass the last `cursor` you have as `after`.

```python
for event in client.tasks.stream(task_id):
    print(event["type"], event["message"])
```

### Everything else

```python
client.tasks.retrieve(task_id, include=["events"])    # include: "events" and/or "trace"
client.tasks.list(session_id=None, agent_id=None, status="completed", limit=20, cursor=None)
client.tasks.list(q="billing email")                  # q searches prompts, task ids, session codes and metadata
client.tasks.list_all(session_id=session_id)          # generator over every page
client.tasks.respond(task_id, question_id="q_...", answer="Work")
client.tasks.respond(task_id, approval_id="appr_...", decision="deny", note="Not on a Friday")
client.tasks.stop(task_id)                            # task["interrupted"] is False if it had already finished

recording = client.tasks.recording(task_id)           # {"frames": [{"seq", "t_ms", "after_event", "width", "height", "url"}]}
png = client.tasks.recording_frame(task_id, recording["frames"][0]["seq"])   # bytes
open("frame-1.png", "wb").write(png)
```

`task["replay_url"]` opens the replay in the console: every screen the agent saw, with its cursor and clicks.

## Pagination

Every list returns one page of up to `limit` items (1 to 100, default 20). Agents, an agent's versions and files, and runners come whole when you pass neither `limit` nor `cursor`, as they did before they paged. Most lists are newest first; agents come `default` first, then by name, an agent's files by path, and runners GuidingHand's first.

```python
page = client.tasks.list(limit=50)
page["data"], page["has_more"], page["next_cursor"]
next_page = client.tasks.list(limit=50, cursor=page["next_cursor"])

page = client.tasks.list(q="printer", include="total_count")
page["total_count"]   # how many match, across every page
```

Every resource with a list also has `list_all(...)` (for a run's attempts, `test_runs.list_all_attempts(...)`). It takes the same filters and `q`, and fetches pages as you iterate; `limit=` sets the page size.

`q` searches a list's text, ignoring case: an agent's name, id and display name; a session's code, computer name and metadata; a task's prompt, id, session code and metadata; a file's path; an audit event's action, details and who did it; a test case's name, key and prompt; a test set's name and description; a start state's name, description and id; a test run's or runner's name and id. `include="total_count"` adds `total_count`. On `test_cases.list`, `include="facets"` adds every category and tag in use, with counts (`include=["total_count", "facets"]` for both).

## Audit log

Who did what in the org: people, API keys, agents (their file writes) and GuidingHand itself. It needs an API key or the admin role.

```python
for e in client.audit_events.list_all(object_type="agent", object_id="billing"):   # an agent's history includes its files'
    print(e["at"], e["action"], e["actor"]["type"])
client.audit_events.list(actor_type="api_key", limit=50)   # one page, newest first
client.audit_events.list(q="jane@acme.com")                # q searches actions, what they were about and who did them
```

See [Audit log](https://docs.guidinghand.ai/concepts/audit-log) for the actions and actors.

## Webhooks

Instead of polling, GuidingHand can POST events to your endpoint: `session.connected`, `session.disconnected`, `task.started`, `task.waiting_for_user`, `task.question_answered`, `task.waiting_for_approval`, `task.approval_decided`, `task.completed`, `task.failed` and `task.stopped`, and for evals `test_run.completed`, `start_state.ready` and `start_state.failed`.

```python
endpoint = client.webhook.update("https://example.com/guidinghand", events=["task.completed", "task.failed"])
secret = endpoint["secret"]   # whsec_...: shown when it is first made, never again. Store it.

client.webhook.retrieve()                              # {"url", "events", "has_secret", "event_types"}
client.webhook.update(events=[])                       # every event; the URL stays
client.webhook.update(rotate_secret=True)["secret"]    # a new secret; the URL and events stay
client.webhook.delete()                                # removes the endpoint and its secret
```

`update` changes only what you pass: leave out `url` or `events` to keep them (the first call needs a `url`). `events` limits which events are sent, and `[]` means all of them (a new endpoint starts with all). An unknown event type raises `InvalidRequestError`. The URL must be a public `https://` address.

Each delivery carries a `GuidingHand-Signature: t=<unix>,v1=<hex>` header. Verify it against the **raw** request body before trusting the event:

```python
from guidinghand import verify_webhook, WebhookVerificationError

# Flask
@app.post("/guidinghand")
def guidinghand_webhook():
    try:
        event = verify_webhook(request.get_data(), request.headers.get("GuidingHand-Signature"), WEBHOOK_SECRET)
    except WebhookVerificationError:
        return "", 400
    if event["type"] == "task.completed":
        task = event["data"]["task"]
        print(task["task_id"], task["result"])
    elif event["type"] == "task.question_answered":
        answer = event["data"]["answer"]   # {"question_id", "answer", "answered_by": "customer" | "operator"}
        print(answer["answered_by"], "answered:", answer["answer"])
    elif event["type"] == "task.approval_decided":
        decision = event["data"]["decision"]   # {"approval_id", "decision", "note", "answered_by": "customer" | "operator" | "timeout"}
        print(decision["answered_by"], "decided:", decision["decision"])
    return "", 200
```

`verify_webhook(payload, signature_header, secret, tolerance=300)` compares signatures in constant time. It rejects timestamps more than `tolerance` seconds from now (`tolerance=None` turns that check off) and returns the parsed event: `{"id", "type", "created_at", "org_id", "data": {"task": ...} | {"session": ...}}` (plus `data["answer"]` for `task.question_answered`, `data["decision"]` for `task.approval_decided`, `data["test_run"]` for `test_run.completed` and `data["start_state"]` for `start_state.ready` and `start_state.failed`). Deliveries that don't get a 2xx are retried up to 4 times over about 3 minutes, so use `event["id"]` to ignore repeats.

## Evals

Measure how often an agent fixes a problem: a **test case** is a problem on a fresh machine and a way to tell it's fixed (a check script that prints `True`, a rubric a model judges, or both). Group cases into a **test set**, **run** it against your agents with a few tries each, and read the pass rates. See [Evals](https://docs.guidinghand.ai/guides/evals).

```python
client.test_cases.create(
    "Apps switched to dark mode",
    "All my windows suddenly went black. How do I get the white back?",   # what the customer says
    "win10-22h2",                                      # the start state each try's machine is made from
    {"check_script": "... -eq 1"},                     # prints True once fixed (and False before the task)
    key="DSP-01",                                      # your own id: every test_cases method takes it
    setup_script="Set-ItemProperty ...",               # breaks the machine at the start of every try
    customer={"persona": "Retired teacher. Not technical."},   # the simulated customer who answers the agent
)
test_set = client.test_sets.create("Windows 10 basics", repeat=3, cases=[{"test_case_id": "DSP-01"}])
client.test_cases.batch(benchmark, test_set_id=test_set["test_set_id"])   # up to 500, created or updated by key
client.test_cases.list(q="dark", include="facets")["facets"]   # {"categories": [{"name", "count"}], "tags": [...]}

run = client.test_sets.run(test_set["test_set_id"], agents=["default", "windows-support"])
done = client.test_runs.wait(run["test_run_id"])       # until completed or cancelled (timeout=, poll_interval= in s)
print(done["pass_rate"])                               # {"default": 0.67, "windows-support": 1.0}
results = client.test_runs.results(run["test_run_id"]) # a row per case and agent; by_agent, by_tag, by_category
for a in client.test_runs.list_all_attempts(run["test_run_id"], outcome="failed"):
    print(a["key"], a["try"], a["check"], a["judge"], a["replay_url"])
client.test_runs.rerun(run["test_run_id"], only="error")   # or "failed" or "all"
```

Start states are the machines tries start from: an image like `win10-22h2`, or one you build by script, by instruction (a GuidingHand task sets it up), or by hand. Runners are the processes that host the machines: GuidingHand's, or your own.

```python
state = client.start_states.create("Printer offline", "win10-22h2", script="Stop-Service Spooler")  # from_ is sent as "from"
state = client.start_states.wait(state["start_state_id"])   # until "ready" or "failed"

s = client.start_states.create("Signed in to Acme", "win10-22h2")   # by hand: it stays "open"
png = client.start_states.screen(s["start_state_id"])       # bytes, admin prompts included
client.start_states.input(s["start_state_id"], [
    {"type": "click", "x": 640, "y": 400},
    {"type": "type", "text": "me@acme.com"},
    {"type": "key", "keys": ["ENTER"]},
])
client.start_states.save(s["start_state_id"])               # or close() to discard

token = client.runners.create("lab-1")["token"]             # your own runner; the token is shown once
```

`AsyncGuidingHand` has the same `test_cases`, `test_sets`, `test_runs`, `attempts`, `start_states` and `runners`, with `await` (and `async for` over `list_all` and `list_all_attempts`).

## Errors

Every exception is a `GuidingHandError`, with `message`, `status` (the HTTP status, or `None`), `type` (the API's error type), `body` (the parsed response) and `extra` (the error's other fields). Exceptions can be pickled, e.g. to pass them between processes.

| Exception | When |
|---|---|
| `InvalidRequestError` | 400: a missing or malformed field |
| `AuthenticationError` | 401: missing or invalid API key (also raised by the constructor when there is none) |
| `PaymentRequiredError` | 402: the org's balance can't pay for the task (`extra["code"]` is `insufficient_balance`, `auto_topup_limit_reached` or `card_needs_attention`) |
| `PermissionDeniedError` | 403: the key's role can't do this |
| `NotFoundError` | 404: no such agent, session, task, frame or eval object in this org |
| `ConflictError` | 409: no computer connected, a task already running, nothing pending to answer, or already answered or decided by someone else (`extra["code"] == "already_answered"`, `extra["answered_by"]`), or a start state that test cases still use (`extra["code"] == "in_use"`, `extra["test_case_ids"]`) |
| `RateLimitError` | 429: too many requests, or too many tasks running at once in the org (`extra["code"] == "concurrency_limit"`) |
| `APIError` | 5xx: a problem on GuidingHand's side, or 503 with `extra["code"] == "model_unavailable"` when the agent's model can't run right now (choose another `model` for the agent) |
| `APIConnectionError` | no response: network failure (`type` is `connection`) or request timeout (`type` is `timeout`) |
| `GuidingHandTimeout` | `wait_for_connection`, `tasks.run(timeout=...)`, `test_runs.wait` or `start_states.wait` ran out of time (`run` stops the task first; if it had just finished, `run` returns it instead; `e.test_run` and `e.start_state` hold the last state) |
| `NeedsInput` | `tasks.run` reached a question or approval (that the customer can't answer or decide) it has no handler for (`e.task`, `e.pending`) |
| `SessionExpired` | the session's code expired while waiting for a connection (`e.session`) |
| `WebhookVerificationError` | a webhook's signature is missing, wrong or too old |

```python
from guidinghand import ConflictError

try:
    client.tasks.create(session_id, "Turn on Dark Mode")
except ConflictError as e:
    running = e.extra.get("active_task_id")   # set when another task is running
    if running:
        client.tasks.stop(running)
```

### Retries

Requests that are safe to repeat are retried up to `max_retries` times (default 2) on connection errors, 429 and 5xx. The waits grow 0.5 s, 1 s, 2 s, and so on, and a `Retry-After` header takes precedence. Safe requests are `GET` and `DELETE`, plus `POST`s that carry a `request_id` (starting a task with `request_id` is idempotent). Other `POST`s, `PATCH` and `PUT` are never retried. To make starting a task safe to retry, pass a `request_id`. If a `DELETE`'s answer is lost and the retry finds the session or agent already gone, the delete returns its usual `{"deleted": True}` result.

## Async

`AsyncGuidingHand` mirrors `GuidingHand`. `stream` and `list_all` are async iterators, and `tasks.run` accepts both plain functions and coroutine functions as handlers.

```python
import asyncio
from guidinghand import AsyncGuidingHand

async def main():
    async with AsyncGuidingHand() as client:
        session = await client.sessions.create()
        print(session["invite_url"])
        await client.sessions.wait_for_connection(session["session_id"])

        async def on_question(q):
            return await ask_support_agent(q["question"], q["options"])

        task = await client.tasks.run(session["session_id"], "Turn on Dark Mode",
                                      on_question=on_question, on_approval=lambda a: a["risk"] == "low")
        async for event in client.tasks.stream(task["task_id"]):
            print(event["type"])

asyncio.run(main())
```

## Development

The tests start the real server (`node server/src/index.js` with an in-memory database), a fake OpenAI, a webhook receiver and fake computers on the WebSocket bridge. They need Node 18+ and `server/node_modules`.

```bash
cd sdks/python
python3 -m venv .venv && .venv/bin/pip install httpx pytest build
.venv/bin/python -m pytest          # or: python3 tests/test_e2e.py
.venv/bin/python -m build           # dist/guidinghand-0.6.0-py3-none-any.whl and .tar.gz
```

Set `NODE=/path/to/node` to choose the Node binary.

## License

MIT
