Python

pip install guidinghand

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"])

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

with GuidingHand() as client:
    ...

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

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)

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

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

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"]),
)

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"])

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

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)

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"])

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

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

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)

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

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