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

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

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:

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 (on by default), and can call your HTTP tools:

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:

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

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 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
    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 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. answered_by is "timeout" when nobody decided within the agent's approval_timeout_minutes, so it was denied.

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

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.

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

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

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.

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.

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

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

License

MIT