Run AI tasks: live view and complete programs

Part of the Run AI tasks from your code. The programs below send an Idempotency-Key; Idempotency keys says which calls take one.

Watch it live (optional)

Step by step, from an SDK: pass onStep to message() to get each step as it finishes, and onEvent for the other progress events (on_step / on_event in Python, OnStep / OnEvent in Go). In TypeScript:

const reply = await client.agentSessions.message(sessionId, task, {
  idempotencyKey: randomUUID(),
  onStep: ({ index, result }) => console.log(`step ${index + 1}: ${result.kind}`),
  onEvent: ({ type, data }) => {
    if (type === 'step_start') console.log('now:', (data as { label: string }).label);
  },
});

In any language: the streamed response carries these events before the final one. Ignore event names you do not recognise — new ones are added over time — and treat response as the only result:

`event`	`data`
`phase`	`{ phase }` — `planning`, `starting_browser`, `executing`, `reading_page` or `answering`
`plan`	`{ total, intents, labels }` — the steps it is about to run
`step_start`	`{ index, total, label }` — a step is starting, e.g. “Opening portal.example.com”
`step`	`{ index, result }` — a step finished
`answer`	`{ answer }`
`challenge_detected`	`{ challenge_id, type, frame_index?, rect? }` — only on accounts with the challenge hand-off on: a verification check paused the turn’s first round of steps and the rest are kept for a “continue” (a check later in the turn sends only `notice`)
`challenge_resume`	`{ replayed, reason? }` — the same accounts: a “continue” after that pause, and whether the kept steps ran (`reason` `steps_gone` when they are no longer held)
`notice`	`{ notice, notice_reason }`
`response`	`{ status, body }` — the result; always last

The whole conversation: transcript(id) in TypeScript and Python, Transcript(ctx, id, opts, fn) in Go, reads every entry the session has written, oldest first (entries dropped from its live history included), and then follows it live over the transcript stream (GET /v1/agent-sessions/{id}/transcript). It does not end by itself while the session is open, so leave the loop when you have what you need — that closes the connection. To read only what is there now, take transcript_length from get(id) first and stop at index === transcript_length - 1:
```
const { transcript_length } = await client.agentSessions.get(sessionId);
if (transcript_length > 0) {
  for await (const { index, entry } of client.agentSessions.transcript(sessionId)) {
    console.log(index, entry.role, entry.body);
    if (index === transcript_length - 1) break;
  }
}
```
Pass the last index you saw back as lastEventId (last_event_id, LastEventID) to carry on where you left off. An account may hold 10 transcript streams open at once, and at most 4 of them may be sending a rolled-off history at the same moment (16 on one server); past either the call is refused with a 429 that says how long to wait. On a long session one call may stop before the end with a TransportError: each SDK holds one stream response to 8 MiB, and the server closes a stream whose reader fell behind (a transcript.error event). Call it again with lastEventId set to the last index you received: nothing is lost or repeated.
Watch the screen: see live video, or open the session in the desktop app.

Complete examples

Each program starts a session on your profile, waits until it is ready, sends the task, answers one question if the agent asks, leaves a step the agent stopped on for a person to approve (or approves it when you set APPROVE_ACTIONS=yes), stops the task after ten minutes, prints the steps and the answer, and always closes the session.

The option names each SDK uses:

Option	TypeScript	Python	Go
Idempotency key for a message	`idempotencyKey`	`idempotency_key`	`IdempotencyKey`
Approve a held step	`approveConsequentialActions`	`approve_consequential_actions`	`ApproveConsequentialActions`
Your own Anthropic key	`byokApiKey`	`byok_api_key`	`ByokAPIKey`
Each step as it finishes	`onStep`	`on_step`	`OnStep`
Other progress events	`onEvent`	`on_event`	`OnEvent`

TypeScript

// run-invoice-task.ts — ask the AI to read this month's invoice total from a supplier portal.
// Run: DRIFTSTACK_API_KEY=ds_live_... PROFILE_ID=prof_... PROXY_ID=... npx tsx run-invoice-task.ts
import { randomUUID } from 'node:crypto';
import {
  Driftstack,
  ConflictError,
  ProfileInUseError,
  RateLimitError,
  TransportError,
  type AgentIntentResult,
  type AgentMessageResponse,
  type ConsequentialActionCategory,
} from '@driftstack/sdk';

const client = new Driftstack({ apiKey: process.env.DRIFTSTACK_API_KEY! });

const PROFILE_ID = process.env.PROFILE_ID!; // a profile you signed in to once, by hand
const PROXY_ID = process.env.PROXY_ID!; // the saved proxy every agent session goes out through
const TASK =
  'Open https://portal.example.com/invoices, find the invoice for September 2026 ' +
  'and tell me its total amount.';
// What the job answers if the agent asks a question part-way.
const REPLY_TO_QUESTIONS =
  'Use the invoice dated September 2026 and tell me its total. Do not pay anything.';
// Leave this off for an unattended job: a person should approve payments.
const APPROVE_ACTIONS = process.env.APPROVE_ACTIONS === 'yes';
const STOP_AFTER_MS = 10 * 60_000;

type Approval = { category: ConsequentialActionCategory; matchedText: string };

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));

// `status` reads `active` from the start; `ready` says when the browser can work.
// Still not ready after two minutes: send anyway, the server no longer holds the task.
async function waitUntilReady(id: string): Promise<void> {
  for (let i = 0; i < 60; i++) {
    const session = await client.agentSessions.get(id);
    if (session.status === 'closed') {
      throw new Error(`The session ended before it was ready: ${session.closed_reason}`);
    }
    if (session.ready !== false) return; // an older server leaves out `ready`: read it as ready
    await sleep(2_000);
  }
}

// One message = one Idempotency-Key. Retrying with the same key never runs the task twice.
async function send(
  id: string,
  text: string,
  approvals: Approval[] = [],
): Promise<AgentMessageResponse> {
  const idempotencyKey = randomUUID();
  // If the task runs too long, ask it to stop; this message then returns kind "stopped".
  const stopTimer = setTimeout(() => {
    client.agentSessions.stop(id).catch(() => undefined);
  }, STOP_AFTER_MS);
  try {
    for (let attempt = 1; ; attempt++) {
      try {
        return await client.agentSessions.message(id, text, {
          idempotencyKey,
          ...(approvals.length > 0 ? { approveConsequentialActions: approvals } : {}),
          onStep: ({ index, result }) => console.log(`  step ${index + 1}: ${describe(result)}`),
        });
      } catch (err) {
        // No response at all, or the first attempt is still running:
        // ask again with the SAME key. Any other error is final for this key.
        const stillRunning =
          err instanceof ConflictError && err.extensions['idempotency_status'] === 'in_progress';
        if ((err instanceof TransportError || stillRunning) && attempt < 6) {
          await sleep(5_000 * attempt);
          continue;
        }
        throw err;
      }
    }
  } finally {
    clearTimeout(stopTimer);
  }
}

function describe(result: AgentIntentResult): string {
  switch (result.kind) {
    case 'success':
      return result.captureId === undefined
        ? `done: ${result.summary}`
        : `done: ${result.summary} (screenshot ${result.captureId})`;
    case 'failure':
      return `failed: ${result.reason} (${result.diagnosis?.category ?? 'unknown'})`;
    case 'confirmation_required':
      return `waiting for approval: ${result.category} "${result.matchedText}"`;
  }
}

function report(reply: AgentMessageResponse): void {
  switch (reply.kind) {
    case 'plan-executed':
      if (reply.notice !== undefined) {
        // `notice_reason` is what an unattended job branches on; the sentence
        // is what a person reads. See "Read what came back".
        console.log(`Not finished (${reply.notice_reason ?? 'no reason given'}): ${reply.notice}`);
      } else if (!reply.ok)
        console.log('A step failed or is waiting for approval; see the steps above.');
      // `answer` and `answer_unavailable` never arrive together, and a task
      // that only acts has neither. Without the second branch a turn that ran
      // cleanly but could not read the page back prints nothing at all.
      if (reply.answer !== undefined) console.log(`Answer: ${reply.answer}`);
      else if (reply.answer_unavailable !== undefined)
        console.log(`No answer: ${reply.answer_unavailable}`);
      break;
    case 'clarify':
      console.log(`The agent still has a question: ${reply.clarifying_question}`);
      break;
    case 'refuse':
      console.log(`Refused: ${reply.refuse_reason}`);
      break;
    case 'stopped':
      console.log(`Stopped while ${reply.stopped_during}: ${reply.notice}`);
      break;
    default:
      console.log(`Unexpected result: ${reply.kind}`);
  }
}

async function main(): Promise<void> {
  const session = await client.agentSessions.create(
    { mode: 'ai', profile_id: PROFILE_ID, proxy_id: PROXY_ID, token_budget: 100_000 },
    { idempotencyKey: randomUUID() },
  );
  try {
    await waitUntilReady(session.id);
    let reply = await send(session.id, TASK);
    const approvals: Approval[] = [];
    for (let followUps = 0; followUps < 3; followUps++) {
      if (reply.kind === 'clarify') {
        console.log(`The agent asked: ${reply.clarifying_question}`);
        reply = await send(session.id, REPLY_TO_QUESTIONS);
        continue;
      }
      const last = reply.kind === 'plan-executed' ? reply.results.at(-1) : undefined;
      if (last?.kind === 'confirmation_required') {
        if (!APPROVE_ACTIONS) {
          console.log(`A person must approve: ${last.category} "${last.matchedText}"`);
          break;
        }
        // Approve by sending the same task again, as the very next message.
        approvals.push({ category: last.category, matchedText: last.matchedText });
        reply = await send(session.id, TASK, approvals);
        continue;
      }
      break;
    }
    report(reply);
  } finally {
    // Always close: it frees the slot and saves the profile's sign-in.
    await client.agentSessions.close(session.id);
  }
}

main().catch((err: unknown) => {
  if (err instanceof ProfileInUseError) {
    console.error(
      `The profile is in use by ${err.activeSessionId}; close it only if you started it.`,
    );
  } else if (err instanceof RateLimitError) {
    console.error(`Too many requests; try again in ${err.retryAfterSeconds}s.`);
  } else {
    console.error(err);
  }
  process.exitCode = 1;
});

Python

# run_invoice_task.py — ask the AI to read this month's invoice total from a supplier portal.
# Run: DRIFTSTACK_API_KEY=ds_live_... PROFILE_ID=prof_... PROXY_ID=... python run_invoice_task.py
import os
import threading
import time
import uuid
from typing import Any

from driftstack import ConflictError, Driftstack, ProfileInUseError, RateLimitError, TransportError

API_KEY = os.environ["DRIFTSTACK_API_KEY"]
PROFILE_ID = os.environ["PROFILE_ID"]  # a profile you signed in to once, by hand
PROXY_ID = os.environ["PROXY_ID"]  # the saved proxy every agent session goes out through
TASK = (
    "Open https://portal.example.com/invoices, find the invoice for September 2026 "
    "and tell me its total amount."
)
# What the job answers if the agent asks a question part-way.
REPLY_TO_QUESTIONS = (
    "Use the invoice dated September 2026 and tell me its total. Do not pay anything."
)
# Leave this off for an unattended job: a person should approve payments.
APPROVE_ACTIONS = os.environ.get("APPROVE_ACTIONS") == "yes"
STOP_AFTER_SECONDS = 10 * 60

client = Driftstack(api_key=API_KEY)


def wait_until_ready(session_id: str) -> None:
    """Wait for `ready`: `status` reads "active" from the start.

    Still not ready after two minutes: send anyway, the server no longer holds the task.
    """
    for _ in range(60):
        session = client.agent_sessions.get(session_id)
        if session["status"] == "closed":
            raise RuntimeError(f"The session ended before it was ready: {session['closed_reason']}")
        if session.get("ready") is not False:  # an older server leaves it out: ready
            return
        time.sleep(2)


def stop_turn(session_id: str) -> None:
    # Runs on a timer thread, so it uses a client of its own.
    with Driftstack(api_key=API_KEY) as timer_client:
        timer_client.agent_sessions.stop(session_id)


def describe(result: dict[str, Any]) -> str:
    if result["kind"] == "success":
        shot = f" (screenshot {result['captureId']})" if "captureId" in result else ""
        return f"done: {result['summary']}{shot}"
    if result["kind"] == "failure":
        category = (result.get("diagnosis") or {}).get("category", "unknown")
        return f"failed: {result['reason']} ({category})"
    if result["kind"] == "confirmation_required":
        return f'waiting for approval: {result["category"]} "{result["matchedText"]}"'
    return str(result["kind"])


def show_step(step: dict[str, Any]) -> None:
    print(f"  step {step['index'] + 1}: {describe(step['result'])}")


def send(
    session_id: str, text: str, approvals: list[dict[str, Any]] | None = None
) -> dict[str, Any]:
    """One message = one Idempotency-Key. Retrying with the same key never runs the task twice."""
    idempotency_key = str(uuid.uuid4())
    # If the task runs too long, ask it to stop; this message then returns kind "stopped".
    stop_timer = threading.Timer(STOP_AFTER_SECONDS, stop_turn, args=(session_id,))
    stop_timer.start()
    try:
        for attempt in range(1, 7):
            try:
                return client.agent_sessions.message(
                    session_id,
                    text,
                    idempotency_key=idempotency_key,
                    approve_consequential_actions=approvals,
                    on_step=show_step,
                )
            except (TransportError, ConflictError) as err:
                # No response at all, or the first attempt is still running:
                # ask again with the SAME key. Any other error is final for this key.
                still_running = (
                    isinstance(err, ConflictError)
                    and err.problem.get("idempotency_status") == "in_progress"
                )
                if (isinstance(err, TransportError) or still_running) and attempt < 6:
                    time.sleep(5 * attempt)
                    continue
                raise
        raise RuntimeError("unreachable")
    finally:
        stop_timer.cancel()


def report(reply: dict[str, Any]) -> None:
    kind = reply["kind"]
    if kind == "plan-executed":
        if "notice" in reply:
            # `notice_reason` is what an unattended job branches on; the
            # sentence is what a person reads. See "Read what came back".
            reason = reply.get("notice_reason", "no reason given")
            print(f"Not finished ({reason}): {reply['notice']}")
        elif not reply["ok"]:
            print("A step failed or is waiting for approval; see the steps above.")
        # `answer` and `answer_unavailable` never arrive together, and a task
        # that only acts has neither. Without the second branch a turn that ran
        # cleanly but could not read the page back prints nothing at all.
        if "answer" in reply:
            print(f"Answer: {reply['answer']}")
        elif "answer_unavailable" in reply:
            print(f"No answer: {reply['answer_unavailable']}")
    elif kind == "clarify":
        print(f"The agent still has a question: {reply['clarifying_question']}")
    elif kind == "refuse":
        print(f"Refused: {reply['refuse_reason']}")
    elif kind == "stopped":
        print(f"Stopped while {reply['stopped_during']}: {reply['notice']}")
    else:
        print(f"Unexpected result: {kind}")


def main() -> None:
    session = client.agent_sessions.create(
        {"mode": "ai", "profile_id": PROFILE_ID, "proxy_id": PROXY_ID, "token_budget": 100_000},
        idempotency_key=str(uuid.uuid4()),
    )
    session_id = session["id"]
    try:
        wait_until_ready(session_id)
        reply = send(session_id, TASK)
        approvals: list[dict[str, Any]] = []
        for _ in range(3):
            if reply["kind"] == "clarify":
                print(f"The agent asked: {reply['clarifying_question']}")
                reply = send(session_id, REPLY_TO_QUESTIONS)
                continue
            results = reply.get("results") or []
            last = results[-1] if reply["kind"] == "plan-executed" and results else None
            if last is not None and last["kind"] == "confirmation_required":
                if not APPROVE_ACTIONS:
                    print(f'A person must approve: {last["category"]} "{last["matchedText"]}"')
                    break
                # Approve by sending the same task again, as the very next message.
                # Pass the held step back as it came: the SDK sends its matchedText
                # as the request's matched_text.
                approvals.append(last)
                reply = send(session_id, TASK, approvals)
                continue
            break
        report(reply)
    finally:
        # Always close: it frees the slot and saves the profile's sign-in.
        client.agent_sessions.close(session_id)


if __name__ == "__main__":
    try:
        main()
    except ProfileInUseError as err:
        raise SystemExit(
            f"The profile is in use by {err.active_session_id}; close it only if you started it."
        ) from err
    except RateLimitError as err:
        raise SystemExit(f"Too many requests; try again in {err.retry_after_seconds}s.") from err

Go

// run-invoice-task — ask the AI to read this month's invoice total from a supplier portal.
// Run: DRIFTSTACK_API_KEY=ds_live_... PROFILE_ID=prof_... PROXY_ID=... go run .
package main

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log"
	"os"
	"time"

	driftstack "github.com/driftstackdev/driftstack-go"
)

const (
	task = "Open https://portal.example.com/invoices, find the invoice for September 2026 " +
		"and tell me its total amount."
	// What the job answers if the agent asks a question part-way.
	replyToQuestions = "Use the invoice dated September 2026 and tell me its total. Do not pay anything."
	stopAfter        = 10 * time.Minute
)

// Leave this off for an unattended job: a person should approve payments.
var approveActions = os.Getenv("APPROVE_ACTIONS") == "yes"

func newKey() string {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		panic(err)
	}
	return hex.EncodeToString(b)
}

// waitUntilReady polls until the session's browser is ready. Status reads
// "active" from the start, so it reads IsReady. Still not ready after two
// minutes: send anyway, the server no longer holds the task.
func waitUntilReady(ctx context.Context, client *driftstack.Client, id string) error {
	for i := 0; i < 60; i++ {
		session, err := client.AgentSessions.Get(ctx, id)
		if err != nil {
			return err
		}
		if session.Status == "closed" {
			reason := ""
			if session.ClosedReason != nil {
				reason = *session.ClosedReason
			}
			return fmt.Errorf("the session ended before it was ready: %s", reason)
		}
		if session.IsReady() {
			return nil
		}
		time.Sleep(2 * time.Second)
	}
	return nil
}

func describe(r driftstack.AgentIntentResult) string {
	switch r.Kind {
	case "success":
		if r.CaptureID != "" {
			return fmt.Sprintf("done: %s (screenshot %s)", r.Summary, r.CaptureID)
		}
		return "done: " + r.Summary
	case "failure":
		category := "unknown"
		if r.Diagnosis != nil {
			category = r.Diagnosis.Category
		}
		return fmt.Sprintf("failed: %s (%s)", r.Reason, category)
	case "confirmation_required":
		return fmt.Sprintf("waiting for approval: %s %q", r.Category, r.MatchedText)
	default:
		return r.Kind
	}
}

// send runs one message. One message = one Idempotency-Key: retrying with the
// same key never runs the task twice.
func send(ctx context.Context, client *driftstack.Client, id, text string,
	approvals []driftstack.ConsequentialActionApproval) (*driftstack.AgentMessageResponse, error) {
	key := newKey()
	// If the task runs too long, ask it to stop; this message then returns kind "stopped".
	stopTimer := time.AfterFunc(stopAfter, func() {
		_, _ = client.AgentSessions.Stop(context.Background(), id)
	})
	defer stopTimer.Stop()
	for attempt := 1; ; attempt++ {
		reply, err := client.AgentSessions.Message(ctx, id, text, &driftstack.MessageOptions{
			IdempotencyKey:              key,
			ApproveConsequentialActions: approvals,
			OnStep: func(step driftstack.AgentStepEvent) {
				fmt.Printf("  step %d: %s\n", step.Index+1, describe(step.Result))
			},
		})
		if err == nil {
			return reply, nil
		}
		// No response at all, or the first attempt is still running:
		// ask again with the SAME key. Any other error is final for this key.
		var conflict *driftstack.ConflictError
		stillRunning := errors.As(err, &conflict) && conflict.Problem["idempotency_status"] == "in_progress"
		if (errors.Is(err, driftstack.ErrTransport) || stillRunning) && attempt < 6 {
			time.Sleep(time.Duration(5*attempt) * time.Second)
			continue
		}
		return nil, err
	}
}

func report(reply *driftstack.AgentMessageResponse) {
	switch reply.Kind {
	case "plan-executed":
		if reply.Notice != "" {
			// NoticeReason is what an unattended job switches on; the sentence
			// is what a person reads. See "Read what came back".
			reason := reply.NoticeReason
			if reason == "" {
				reason = "no reason given"
			}
			fmt.Printf("Not finished (%s): %s\n", reason, reply.Notice)
		} else if !reply.OK {
			fmt.Println("A step failed or is waiting for approval; see the steps above.")
		}
		// Answer and AnswerUnavailable never arrive together, and a task that
		// only acts has neither. Without the second branch a turn that ran
		// cleanly but could not read the page back prints nothing at all.
		if reply.Answer != "" {
			fmt.Println("Answer:", reply.Answer)
		} else if reply.AnswerUnavailable != "" {
			fmt.Println("No answer:", reply.AnswerUnavailable)
		}
	case "clarify":
		fmt.Println("The agent still has a question:", reply.ClarifyingQuestion)
	case "refuse":
		fmt.Println("Refused:", reply.RefuseReason)
	case "stopped":
		fmt.Printf("Stopped while %s: %s\n", reply.StoppedDuring, reply.Notice)
	default:
		fmt.Println("Unexpected result:", reply.Kind)
	}
}

func run(ctx context.Context, client *driftstack.Client) error {
	session, err := client.AgentSessions.Create(ctx, &driftstack.CreateAgentSessionRequest{
		Mode:        "ai",
		ProfileID:   os.Getenv("PROFILE_ID"), // a profile you signed in to once, by hand
		ProxyID:     os.Getenv("PROXY_ID"),   // the saved proxy every agent session goes out through
		TokenBudget: 100000,
	}, &driftstack.CreateOptions{IdempotencyKey: newKey()})
	if err != nil {
		var busy *driftstack.ProfileInUseError
		if errors.As(err, &busy) {
			return fmt.Errorf("the profile is in use by %s; close it only if you started it", busy.ActiveSessionID)
		}
		return err
	}
	// Always close: it frees the slot and saves the profile's sign-in.
	defer func() {
		if err := client.AgentSessions.Close(context.Background(), session.ID); err != nil {
			log.Printf("close failed: %v", err)
		}
	}()

	if err := waitUntilReady(ctx, client, session.ID); err != nil {
		return err
	}
	reply, err := send(ctx, client, session.ID, task, nil)
	if err != nil {
		return err
	}
	var approvals []driftstack.ConsequentialActionApproval
	for followUps := 0; followUps < 3; followUps++ {
		if reply.Kind == "clarify" {
			fmt.Println("The agent asked:", reply.ClarifyingQuestion)
			if reply, err = send(ctx, client, session.ID, replyToQuestions, nil); err != nil {
				return err
			}
			continue
		}
		steps, err := reply.ParsedResults()
		if err != nil {
			return err
		}
		if reply.Kind != "plan-executed" || len(steps) == 0 || steps[len(steps)-1].Kind != "confirmation_required" {
			break
		}
		last := steps[len(steps)-1]
		if !approveActions {
			fmt.Printf("A person must approve: %s %q\n", last.Category, last.MatchedText)
			break
		}
		// Approve by sending the same task again, as the very next message.
		approvals = append(approvals, driftstack.ApprovalFor(last))
		if reply, err = send(ctx, client, session.ID, task, approvals); err != nil {
			return err
		}
	}
	report(reply)
	return nil
}

func main() {
	client := driftstack.New(os.Getenv("DRIFTSTACK_API_KEY"))
	defer client.Close()
	if err := run(context.Background(), client); err != nil {
		var limited *driftstack.RateLimitError
		if errors.As(err, &limited) {
			log.Printf("too many requests; try again in %ds", limited.RetryAfterSeconds)
		}
		log.Print(err)
		os.Exit(1)
	}
}

curl

The same flow with raw HTTP, as a shell script. It needs jq, and uuidgen for the keys.

#!/usr/bin/env bash
# run-invoice-task.sh — ask the AI to read this month's invoice total from a supplier portal.
# Run: DRIFTSTACK_API_KEY=ds_live_... PROFILE_ID=prof_... PROXY_ID=... bash run-invoice-task.sh
: "${DRIFTSTACK_API_KEY:?set DRIFTSTACK_API_KEY to your API key}"
: "${PROFILE_ID:?set PROFILE_ID to the prof_… you signed in by hand}"
: "${PROXY_ID:?set PROXY_ID to the id of your saved proxy}"
API=https://api.driftstack.dev
AUTH="Authorization: Bearer $DRIFTSTACK_API_KEY"
TASK='Open https://portal.example.com/invoices, find the invoice for September 2026 and tell me its total amount.'

# 1. Start a session on the signed-in profile, through your saved proxy. Keep
#    CREATE_KEY to retry the create safely.
CREATE_KEY=$(uuidgen)
jq -n --arg p "$PROFILE_ID" --arg x "$PROXY_ID" \
  '{mode: "ai", profile_id: $p, proxy_id: $x, token_budget: 100000}' > create.json
curl -sS "$API/v1/agent-sessions" -H "$AUTH" -H "Content-Type: application/json" \
  -H "Idempotency-Key: $CREATE_KEY" --data @create.json > session.json
ID=$(jq -r .id session.json)
[ "$ID" != null ] || { cat session.json; exit 1; } # for example 409 profile-in-use or 422 proxy-required

# 2. Wait until its browser is ready. `status` reads `active` from the start, so
#    read `ready`. Still not ready after two minutes: send anyway.
for _ in $(seq 60); do
  curl -sS "$API/v1/agent-sessions/$ID" -H "$AUTH" > state.json
  [ "$(jq -r .id state.json)" = "$ID" ] || { cat state.json; exit 1; } # an error, such as 401 or 404
  [ "$(jq -r .status state.json)" = closed ] && { echo "The session ended before it was ready."; exit 1; }
  [ "$(jq -r .ready state.json)" = true ] && break
  sleep 2
done

# 3. send FILE — one message, streamed, under a new Idempotency-Key. The last event,
#    `response`, carries { status, body }; it is saved to result.json.
send() {
  TURN_KEY=$(uuidgen)
  curl -sS -N --max-time 3000 "$API/v1/agent-sessions/$ID/message" -H "$AUTH" \
    -H "Content-Type: application/json" -H "Accept: text/event-stream" \
    -H "Idempotency-Key: $TURN_KEY" --data @"$1" > turn.sse
  awk '/^event: response$/ { getline; sub(/^data: /, ""); print }' turn.sse > result.json
  # No `response` event: an ordinary HTTP error (bad key, missing scope, too
  # many requests) or a dropped connection. The body says which.
  [ -s result.json ] || { cat turn.sse; exit 1; }
  jq '{status, kind: .body.kind, ok: .body.ok, answer: .body.answer, notice: .body.notice,
       notice_reason: .body.notice_reason, question: .body.clarifying_question,
       steps: [.body.results[]? | .kind]}' result.json
}

jq -n --arg m "$TASK" '{user_message: $m}' > message.json
send message.json

# 4a. It asked a question (kind "clarify"): answer it as the next message.
if [ "$(jq -r .body.kind result.json)" = clarify ]; then
  jq -n '{user_message: "Use the invoice dated September 2026 and tell me its total. Do not pay anything."}' > reply.json
  send reply.json
fi

# 4b. A step is held for approval: approve it (only with APPROVE_ACTIONS=yes) by sending
#     the same task again, as the very next message, with the approval.
HELD=$(jq -r '.body.results[-1].kind // empty' result.json)
if [ "$HELD" = confirmation_required ] && [ "${APPROVE_ACTIONS:-}" = yes ]; then
  jq -n --arg m "$TASK" --slurpfile r result.json '{user_message: $m,
    approve_consequential_actions: [{category: $r[0].body.results[-1].category,
                                     matched_text: $r[0].body.results[-1].matchedText}]}' \
    > approve.json
  send approve.json
fi

# 5. To stop a task that runs too long, run this from another shell while `send` waits;
#    the waiting message then ends with kind "stopped":
#    curl -sS -X POST "$API/v1/agent-sessions/$ID/stop" -H "$AUTH" -H "Content-Type: application/json" -d '{}'

# 6. Always close the session: it frees the slot and saves the profile's sign-in. Prints 204.
curl -sS -X DELETE "$API/v1/agent-sessions/$ID" -H "$AUTH" -o /dev/null -w '%{http_code}\n'

If the curl inside send loses its connection, run it again with the same TURN_KEY: you get the stored result, or a 409 with idempotency_status: "in_progress" while the task is still running.