GuidingHand Python SDK

The official Python client for the GuidingHand 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.

pip install guidinghand

Python 3.9 or later. The only dependency is httpx.

Quickstart

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

export GUIDINGHAND_API_KEY="gh_live_..."

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 TypedDicts (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

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.

with GuidingHand() as client:
    ...

AsyncGuidingHand has the same resources and methods, and you await them. See 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.

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.

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 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.

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 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:

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.

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:

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.

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

Everything else

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:

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.

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:

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.

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 POSTs that carry a request_id (starting a task with request_id is idempotent). Other POSTs, 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.

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.

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.3.0-py3-none-any.whl and .tar.gz

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

License

MIT