Metadata-Version: 2.5
Name: guidinghand
Version: 0.2.0
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, 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": False, "next_cursor": None}
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.",
    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)
)
client.agents.retrieve("billing")
client.agents.update("billing", effort="high")   # only the fields you pass; effort=None resets it
client.agents.update("billing", customer_answers=False)   # only your team answers questions
client.agents.delete("billing")          # its sessions switch to default; deleting "default" resets it
```

`agent["invite_url_template"]` is the agent's invite link with a `{code}` placeholder. Creating an `agent_id` that already exists raises `ConflictError`.

## 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
for s in client.sessions.list_all(agent_id="billing"):        # every page
    print(s["session_id"], s["status"], s.get("task_count"))
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
    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"}
    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. 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 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.

```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: ask_my_team_to_approve(a["action"]),
)
```

Approvals always come from your team.

### 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_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

Lists return one page, newest 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"])
```

`sessions.list_all(...)` and `tasks.list_all(...)` take the same filters and fetch pages as you iterate. They also take `limit=` for the page size.

## 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.completed`, `task.failed` and `task.stopped`.

```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"])
    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`). Deliveries that don't get a 2xx are retried up to 4 times over about 3 minutes, so use `event["id"]` to ignore repeats.

## 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 plan doesn't allow it, e.g. its free minutes are used up |
| `PermissionDeniedError` | 403: the key's role can't do this |
| `NotFoundError` | 404: no such agent, session, task or frame in this org |
| `ConflictError` | 409: no computer connected, a task already running, nothing pending to answer, or already answered by someone else (`extra["code"] == "already_answered"`, `extra["answered_by"]`) |
| `RateLimitError` | 429: too many requests |
| `APIError` | 5xx: a problem on GuidingHand's side |
| `APIConnectionError` | no response: network failure (`type` is `connection`) or request timeout (`type` is `timeout`) |
| `GuidingHandTimeout` | `wait_for_connection` or `tasks.run(timeout=...)` ran out of time (`run` stops the task first; if it had just finished, `run` returns it instead) |
| `NeedsInput` | `tasks.run` reached a question (that the customer can't answer) or an approval 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.2.0-py3-none-any.whl and .tar.gz
```

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

## License

MIT
