> ## Documentation Index > Fetch the complete documentation index at: https://docs.guidinghand.ai/llms.txt > Use this file to discover all available pages before exploring further. # Python > The guidinghand package for Python 3.9+, sync and async. 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 theme={null} 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 theme={null} export GUIDINGHAND_API_KEY="gh_live_..." ``` ```python theme={null} 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 theme={null} 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 theme={null} 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 theme={null} 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) customer_approvals=True, # they can approve or deny its approval requests 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.update("billing", customer_approvals=False) # only your team decides approvals 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 theme={null} 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 theme={null} 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", "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 theme={null} 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. ```python theme={null} 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 theme={null} 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 theme={null} for event in client.tasks.stream(task_id): print(event["type"], event["message"]) ``` ### Everything else ```python theme={null} 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 theme={null} 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.approval_decided`, `task.completed`, `task.failed` and `task.stopped`. ```python theme={null} 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=,v1=` header. Verify it against the **raw** request body before trusting the event: ```python theme={null} 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"} 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` and `data["decision"]` for `task.approval_decided`). 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 or decided 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 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 theme={null} 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 theme={null} 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 theme={null} 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