Approve or deny tool calls

The scripts p3-tool-approvals.sh, p3-tool-approvals.py and p3-tool-approvals.ts are in full under Complete script.

Before you start

Set CONTEXT_BASE_URL to your deployment’s host, without /v1 (for example, https://api.context.ai).
Set CONTEXT_API_KEY to a personal Task API key with tasks:approve. Keys don’t have it by default. The key’s owner can add it only when creating the key.
Set AGENT_ID to an agent with a tool set to Ask in the app, for example bash.

Client setup used by every step

#!/usr/bin/env bash
set -euo pipefail
: "${CONTEXT_BASE_URL:?Set CONTEXT_BASE_URL, for example https://api.context.ai}"
: "${CONTEXT_API_KEY:?Set CONTEXT_API_KEY to a personal Task API key}"
: "${AGENT_ID:?Set AGENT_ID to an agent with a tool set to Ask in the app}"
api() {
  local path=$1; shift
  curl -sS --fail-with-body "$CONTEXT_BASE_URL/v1$path" -H "x-api-key: $CONTEXT_API_KEY" "$@"
}
fail() { echo "error: $*" >&2; exit 1; }
# Opens the session's event stream; exec makes the background job's pid the curl process itself.
stream() {
  exec curl -sS --fail-with-body -N "$CONTEXT_BASE_URL/v1/sessions/$1/events/stream" \
    -H "x-api-key: $CONTEXT_API_KEY" "${@:2}"
}
# The JSON of every event received so far on a stream saved to a file. Ping frames carry no id and are skipped.
events_in() { sed -n 's/^data: //p' "$1" | jq -c 'select(.id)'; }
# Waits for the newest session.status_idle on the stream file and prints its stop_reason JSON.
wait_for_idle() {
  local after=$1 file=$2
  for _ in $(seq 1 300); do
    local idle
    grep -q '^event: error' "$file" && fail "the stream ended with an error: $(grep -A2 '^event: error' "$file" | sed -n 's/^data: //p')"
    idle=$(events_in "$file" | jq -sc --argjson after "$after" '[.[] | select(.type == "session.status_idle")] | .[$after:] | .[0].stop_reason // empty')
    [[ -n $idle ]] && { echo "$idle"; return 0; }
    sleep 1
  done
  fail "no session.status_idle within 300 seconds"
}
WORK=$(mktemp -d)

import os

from anthropic import Anthropic

client = Anthropic(
    base_url=os.environ["CONTEXT_BASE_URL"],
    api_key=os.environ["CONTEXT_API_KEY"],
    max_retries=0,
)
agent_id = os.environ["AGENT_ID"]  # an agent with a tool set to Ask in the app


def check(ok, message):
    if not ok:
        raise SystemExit(f"error: {message}")


def next_idle(events):
    """Reads events until the next session.status_idle; returns its stop_reason and the agent's text."""
    text = ""
    for event in events:
        if event.type == "agent.message":
            text += "".join(block.text for block in event.content if block.type == "text")
        if event.type == "session.status_idle":
            return event.stop_reason, text
    raise SystemExit("error: the stream ended before session.status_idle")

import Anthropic from "@anthropic-ai/sdk"
import type { BetaManagedAgentsStreamSessionEvents } from "@anthropic-ai/sdk/resources/beta/sessions/events"

const client = new Anthropic({
  baseURL: process.env.CONTEXT_BASE_URL,
  apiKey: process.env.CONTEXT_API_KEY,
  maxRetries: 0,
})
const agentId = process.env.AGENT_ID! // an agent with a tool set to Ask in the app

function check(ok: unknown, message: string): asserts ok {
  if (!ok) throw new Error(message)
}
/** Reads events until the next session.status_idle; returns its stop_reason and the agent's text. */
async function nextIdle(events: AsyncIterator<BetaManagedAgentsStreamSessionEvents>) {
  let text = ""
  for (let next = await events.next(); !next.done; next = await events.next()) {
    const event = next.value
    if (event.type === "agent.message") text += event.content.map(b => (b.type === "text" ? b.text : "")).join("")
    if (event.type === "session.status_idle") return { stop: event.stop_reason, text }
  }
  throw new Error("the stream ended before session.status_idle")
}

Create a session and open the stream

Open the stream before the first message so the approval request is not missed.

SESSION_ID=$(api /sessions -H "content-type: application/json" \
  -d "$(jq -n --arg agent "$AGENT_ID" '{agent: $agent, environment_id: "env_default", title: "Approvals demo"}')" | jq -r .id)
[[ -n $SESSION_ID && $SESSION_ID != null ]] || fail "no session id"
stream "$SESSION_ID" > "$WORK/stream.txt" &
STREAM_PID=$!
sleep 1
kill -0 "$STREAM_PID" 2>/dev/null || fail "the stream did not open: $(cat "$WORK/stream.txt")"
echo "step 1: session $SESSION_ID, stream open"

session = client.beta.sessions.create(agent=agent_id, environment_id="env_default", title="Approvals demo")
stream = client.beta.sessions.events.stream(session.id)
events = iter(stream)
check(session.status == "idle", f"expected an idle session, got {session.status}")
print(f"step 1: session {session.id}, stream open")

const session = await client.beta.sessions.create({ agent: agentId, environment_id: "env_default", title: "Approvals demo" })
const stream = await client.beta.sessions.events.stream(session.id)
const events = stream[Symbol.asyncIterator]()
check(session.status === "idle", `expected an idle session, got ${session.status}`)
console.log(`step 1: session ${session.id}, stream open`)

Sample output:

step 1: session dd13d728-93e3-440a-874c-4efeb58c97e0, stream open

Ask for work that needs a tool

When the agent wants a tool that needs approval, the session goes idle with stop_reason.type set to requires_action. stop_reason.event_ids lists the agent.tool_use ids waiting for you.

api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d '{"events": [{"type": "user.message",
  "content": [{"type": "text", "text": "Run this bash command and tell me the output: echo approved"}]}]}' > /dev/null
STOP=$(wait_for_idle 0 "$WORK/stream.txt")
[[ $(jq -r .type <<<"$STOP") == "requires_action" ]] || fail "expected requires_action, got $STOP"
echo "step 2: waiting for approval of $(jq -r '.event_ids | join(", ")' <<<"$STOP")"

client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": "Run this bash command and tell me the output: echo approved"}]}],
)
stop, _ = next_idle(events)
check(stop.type == "requires_action", f"expected requires_action, got {stop.type}")
print(f"step 2: waiting for approval of {', '.join(stop.event_ids)}")

await client.beta.sessions.events.send(session.id, {
  events: [{ type: "user.message", content: [{ type: "text", text: "Run this bash command and tell me the output: echo approved" }] }],
})
const { stop } = await nextIdle(events)
check(stop.type === "requires_action", `expected requires_action, got ${stop.type}`)
console.log(`step 2: waiting for approval of ${stop.event_ids.join(", ")}`)

Sample output:

step 2: waiting for approval of sevt_29_0

Approve each waiting tool call

Send one user.tool_confirmation per id, with that id as tool_use_id. Set result to allow to approve that one call, or to deny to refuse it. With deny, an optional deny_message is stored with the answer and echoed on the event.

for TOOL_USE_ID in $(jq -r '.event_ids[]' <<<"$STOP"); do
  api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d "$(jq -n --arg id "$TOOL_USE_ID" \
    '{events: [{type: "user.tool_confirmation", tool_use_id: $id, result: "allow"}]}')" > /dev/null
  # To refuse: result: "deny", with an optional deny_message that is stored with the answer and echoed on the event
done
echo "step 3: allowed $(jq '.event_ids | length' <<<"$STOP") tool call(s)"

for tool_use_id in stop.event_ids:
    client.beta.sessions.events.send(
        session.id,
        events=[{"type": "user.tool_confirmation", "tool_use_id": tool_use_id, "result": "allow"}],
        # To refuse: "result": "deny", with an optional "deny_message" that is stored with the answer and echoed on the event
    )
print(f"step 3: allowed {len(stop.event_ids)} tool call(s)")

for (const toolUseId of stop.event_ids) {
  await client.beta.sessions.events.send(session.id, {
    events: [{ type: "user.tool_confirmation", tool_use_id: toolUseId, result: "allow" }],
    // To refuse: result: "deny", with an optional deny_message that is stored with the answer and echoed on the event
  })
}
console.log(`step 3: allowed ${stop.event_ids.length} tool call(s)`)

Sample output:

step 3: allowed 1 tool call(s)

Read to the end of the turn

After the last confirmation, the agent runs the tool and continues until end_turn.

STOP=$(wait_for_idle 1 "$WORK/stream.txt")
kill "$STREAM_PID"
[[ $(jq -r .type <<<"$STOP") == "end_turn" ]] || fail "expected end_turn, got $STOP"
ANSWER=$(events_in "$WORK/stream.txt" | jq -r 'select(.type == "agent.message") | .content[] | select(.type == "text") | .text')
echo "step 4: end_turn, agent said: $ANSWER"

stop, answer = next_idle(events)
stream.close()
check(stop.type == "end_turn", f"expected end_turn, got {stop.type}")
print(f"step 4: end_turn, agent said: {answer}")

const end = await nextIdle(events)
stream.controller.abort()
check(end.stop.type === "end_turn", `expected end_turn, got ${end.stop.type}`)
console.log(`step 4: end_turn, agent said: ${end.text}`)

Sample output:

step 4: end_turn, agent said: The command printed: approved

Archive the session

Archiving ends the session. Its status becomes terminated and its history stays readable.

ARCHIVED=$(api "/sessions/$SESSION_ID/archive" -X POST)
[[ $(jq -r .status <<<"$ARCHIVED") == "terminated" ]] || fail "expected terminated: $ARCHIVED"
[[ $(jq -r .archived_at <<<"$ARCHIVED") != null ]] || fail "archived_at is not set: $ARCHIVED"
echo "step 5: session terminated at $(jq -r .archived_at <<<"$ARCHIVED")"

archived = client.beta.sessions.archive(session.id)
check(archived.status == "terminated", f"expected terminated, got {archived.status}")
check(archived.archived_at is not None, "archived_at is not set")
print(f"step 5: session {archived.status} at {archived.archived_at}")

const archived = await client.beta.sessions.archive(session.id)
check(archived.status === "terminated", `expected terminated, got ${archived.status}`)
check(archived.archived_at, "archived_at is not set")
console.log(`step 5: session ${archived.status} at ${archived.archived_at}`)

Sample output:

step 5: session terminated at 2026-10-21T01:47:50.964Z

If something goes wrong

What you see	Why	What to do
The turn ends with `end_turn` and step 2 fails	The agent’s tools do not ask for permission, or the agent answered without a tool.	In the app, set the agent’s `bash` tool to Ask, then run again.
`403 permission_error` on `user.tool_confirmation`	The key lacks `tasks:approve`, or it is a service account key. Only a personal key with `tasks:approve`, owned by the session’s owner, can answer.	Have the session’s owner create a new personal key with `tasks:approve`. For a service account key, see the next row.
A session started with a service account key sits idle with `requires_action`	A service account key cannot answer a tool call, so a tool set to Ask is never approved and the session waits.	In the app, set the tools the agent uses to Allow before you start sessions with that key. To approve tool calls yourself, use a personal key with `tasks:approve`.
`400 invalid_request_error`: `tool_use_id ... has no confirmation waiting`	The id was already answered, or it is not from the latest `stop_reason.event_ids`.	Send exactly the ids from the latest `requires_action`, once each.
`409 invalid_request_error` on `user.tool_confirmation`	Another client answered the tool call first, or the session is stopping or moving to another computer.	List the session’s events, find the current `requires_action` idle and answer its `event_ids`.
The session stays idle after approving	Not every id in `event_ids` was answered.	Answer every id in `stop_reason.event_ids`. The agent resumes after the last one.

Complete script

#!/usr/bin/env bash
set -euo pipefail
: "${CONTEXT_BASE_URL:?Set CONTEXT_BASE_URL, for example https://api.context.ai}"
: "${CONTEXT_API_KEY:?Set CONTEXT_API_KEY to a personal Task API key}"
: "${AGENT_ID:?Set AGENT_ID to an agent with a tool set to Ask in the app}"
api() {
  local path=$1; shift
  curl -sS --fail-with-body "$CONTEXT_BASE_URL/v1$path" -H "x-api-key: $CONTEXT_API_KEY" "$@"
}
fail() { echo "error: $*" >&2; exit 1; }
# Opens the session's event stream; exec makes the background job's pid the curl process itself.
stream() {
  exec curl -sS --fail-with-body -N "$CONTEXT_BASE_URL/v1/sessions/$1/events/stream" \
    -H "x-api-key: $CONTEXT_API_KEY" "${@:2}"
}
# The JSON of every event received so far on a stream saved to a file. Ping frames carry no id and are skipped.
events_in() { sed -n 's/^data: //p' "$1" | jq -c 'select(.id)'; }
# Waits for the newest session.status_idle on the stream file and prints its stop_reason JSON.
wait_for_idle() {
  local after=$1 file=$2
  for _ in $(seq 1 300); do
    local idle
    grep -q '^event: error' "$file" && fail "the stream ended with an error: $(grep -A2 '^event: error' "$file" | sed -n 's/^data: //p')"
    idle=$(events_in "$file" | jq -sc --argjson after "$after" '[.[] | select(.type == "session.status_idle")] | .[$after:] | .[0].stop_reason // empty')
    [[ -n $idle ]] && { echo "$idle"; return 0; }
    sleep 1
  done
  fail "no session.status_idle within 300 seconds"
}
WORK=$(mktemp -d)

# @step 1 Create a session and open the stream
SESSION_ID=$(api /sessions -H "content-type: application/json" \
  -d "$(jq -n --arg agent "$AGENT_ID" '{agent: $agent, environment_id: "env_default", title: "Approvals demo"}')" | jq -r .id)
[[ -n $SESSION_ID && $SESSION_ID != null ]] || fail "no session id"
stream "$SESSION_ID" > "$WORK/stream.txt" &
STREAM_PID=$!
sleep 1
kill -0 "$STREAM_PID" 2>/dev/null || fail "the stream did not open: $(cat "$WORK/stream.txt")"
echo "step 1: session $SESSION_ID, stream open"

# @step 2 Ask for work that needs a tool
api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d '{"events": [{"type": "user.message",
  "content": [{"type": "text", "text": "Run this bash command and tell me the output: echo approved"}]}]}' > /dev/null
STOP=$(wait_for_idle 0 "$WORK/stream.txt")
[[ $(jq -r .type <<<"$STOP") == "requires_action" ]] || fail "expected requires_action, got $STOP"
echo "step 2: waiting for approval of $(jq -r '.event_ids | join(", ")' <<<"$STOP")"

# @step 3 Approve each waiting tool call
for TOOL_USE_ID in $(jq -r '.event_ids[]' <<<"$STOP"); do
  api "/sessions/$SESSION_ID/events" -H "content-type: application/json" -d "$(jq -n --arg id "$TOOL_USE_ID" \
    '{events: [{type: "user.tool_confirmation", tool_use_id: $id, result: "allow"}]}')" > /dev/null
  # To refuse: result: "deny", with an optional deny_message that is stored with the answer and echoed on the event
done
echo "step 3: allowed $(jq '.event_ids | length' <<<"$STOP") tool call(s)"

# @step 4 Read to the end of the turn
STOP=$(wait_for_idle 1 "$WORK/stream.txt")
kill "$STREAM_PID"
[[ $(jq -r .type <<<"$STOP") == "end_turn" ]] || fail "expected end_turn, got $STOP"
ANSWER=$(events_in "$WORK/stream.txt" | jq -r 'select(.type == "agent.message") | .content[] | select(.type == "text") | .text')
echo "step 4: end_turn, agent said: $ANSWER"

# @step 5 Archive the session
ARCHIVED=$(api "/sessions/$SESSION_ID/archive" -X POST)
[[ $(jq -r .status <<<"$ARCHIVED") == "terminated" ]] || fail "expected terminated: $ARCHIVED"
[[ $(jq -r .archived_at <<<"$ARCHIVED") != null ]] || fail "archived_at is not set: $ARCHIVED"
echo "step 5: session terminated at $(jq -r .archived_at <<<"$ARCHIVED")"
# @end

import os

from anthropic import Anthropic

client = Anthropic(
    base_url=os.environ["CONTEXT_BASE_URL"],
    api_key=os.environ["CONTEXT_API_KEY"],
    max_retries=0,
)
agent_id = os.environ["AGENT_ID"]  # an agent with a tool set to Ask in the app


def check(ok, message):
    if not ok:
        raise SystemExit(f"error: {message}")


def next_idle(events):
    """Reads events until the next session.status_idle; returns its stop_reason and the agent's text."""
    text = ""
    for event in events:
        if event.type == "agent.message":
            text += "".join(block.text for block in event.content if block.type == "text")
        if event.type == "session.status_idle":
            return event.stop_reason, text
    raise SystemExit("error: the stream ended before session.status_idle")


# @step 1 Create a session and open the stream
session = client.beta.sessions.create(agent=agent_id, environment_id="env_default", title="Approvals demo")
stream = client.beta.sessions.events.stream(session.id)
events = iter(stream)
check(session.status == "idle", f"expected an idle session, got {session.status}")
print(f"step 1: session {session.id}, stream open")

# @step 2 Ask for work that needs a tool
client.beta.sessions.events.send(
    session.id,
    events=[{"type": "user.message", "content": [{"type": "text", "text": "Run this bash command and tell me the output: echo approved"}]}],
)
stop, _ = next_idle(events)
check(stop.type == "requires_action", f"expected requires_action, got {stop.type}")
print(f"step 2: waiting for approval of {', '.join(stop.event_ids)}")

# @step 3 Approve each waiting tool call
for tool_use_id in stop.event_ids:
    client.beta.sessions.events.send(
        session.id,
        events=[{"type": "user.tool_confirmation", "tool_use_id": tool_use_id, "result": "allow"}],
        # To refuse: "result": "deny", with an optional "deny_message" that is stored with the answer and echoed on the event
    )
print(f"step 3: allowed {len(stop.event_ids)} tool call(s)")

# @step 4 Read to the end of the turn
stop, answer = next_idle(events)
stream.close()
check(stop.type == "end_turn", f"expected end_turn, got {stop.type}")
print(f"step 4: end_turn, agent said: {answer}")

# @step 5 Archive the session
archived = client.beta.sessions.archive(session.id)
check(archived.status == "terminated", f"expected terminated, got {archived.status}")
check(archived.archived_at is not None, "archived_at is not set")
print(f"step 5: session {archived.status} at {archived.archived_at}")
# @end

import Anthropic from "@anthropic-ai/sdk"
import type { BetaManagedAgentsStreamSessionEvents } from "@anthropic-ai/sdk/resources/beta/sessions/events"

const client = new Anthropic({
  baseURL: process.env.CONTEXT_BASE_URL,
  apiKey: process.env.CONTEXT_API_KEY,
  maxRetries: 0,
})
const agentId = process.env.AGENT_ID! // an agent with a tool set to Ask in the app

function check(ok: unknown, message: string): asserts ok {
  if (!ok) throw new Error(message)
}
/** Reads events until the next session.status_idle; returns its stop_reason and the agent's text. */
async function nextIdle(events: AsyncIterator<BetaManagedAgentsStreamSessionEvents>) {
  let text = ""
  for (let next = await events.next(); !next.done; next = await events.next()) {
    const event = next.value
    if (event.type === "agent.message") text += event.content.map(b => (b.type === "text" ? b.text : "")).join("")
    if (event.type === "session.status_idle") return { stop: event.stop_reason, text }
  }
  throw new Error("the stream ended before session.status_idle")
}

// @step 1 Create a session and open the stream
const session = await client.beta.sessions.create({ agent: agentId, environment_id: "env_default", title: "Approvals demo" })
const stream = await client.beta.sessions.events.stream(session.id)
const events = stream[Symbol.asyncIterator]()
check(session.status === "idle", `expected an idle session, got ${session.status}`)
console.log(`step 1: session ${session.id}, stream open`)

// @step 2 Ask for work that needs a tool
await client.beta.sessions.events.send(session.id, {
  events: [{ type: "user.message", content: [{ type: "text", text: "Run this bash command and tell me the output: echo approved" }] }],
})
const { stop } = await nextIdle(events)
check(stop.type === "requires_action", `expected requires_action, got ${stop.type}`)
console.log(`step 2: waiting for approval of ${stop.event_ids.join(", ")}`)

// @step 3 Approve each waiting tool call
for (const toolUseId of stop.event_ids) {
  await client.beta.sessions.events.send(session.id, {
    events: [{ type: "user.tool_confirmation", tool_use_id: toolUseId, result: "allow" }],
    // To refuse: result: "deny", with an optional deny_message that is stored with the answer and echoed on the event
  })
}
console.log(`step 3: allowed ${stop.event_ids.length} tool call(s)`)

// @step 4 Read to the end of the turn
const end = await nextIdle(events)
stream.controller.abort()
check(end.stop.type === "end_turn", `expected end_turn, got ${end.stop.type}`)
console.log(`step 4: end_turn, agent said: ${end.text}`)

// @step 5 Archive the session
const archived = await client.beta.sessions.archive(session.id)
check(archived.status === "terminated", `expected terminated, got ${archived.status}`)
check(archived.archived_at, "archived_at is not set")
console.log(`step 5: session ${archived.status} at ${archived.archived_at}`)
// @end