Progress updates from a Claude Agent SDK agent

What the message stream contains

• SystemMessage with subtype init comes first. It carries session metadata, the tool list and the status of each MCP server.
• AssistantMessage arrives as each content block finishes, so a text block and a tool call are two messages. Read the tool_use block's name to show "Using Bash" or "Using Read".
• UserMessage follows each tool run with the result sent back to Claude.
• StreamEvent in Python, or a message with type stream_event in TypeScript, carries raw API events such as text deltas. You only get these with partial messages turned on.
• ResultMessage ends the run with the final text, a subtype (success, error_max_turns, error_max_budget_usd and others), the number of turns, the duration and total_cost_usd. A few system events can still follow it, so read the stream to the end rather than breaking on the result.

In TypeScript you check message.type, and an assistant message's content blocks are at message.message.content. In Python you check isinstance against classes from claude_agent_sdk, and the blocks are at message.content.

Print each step as it happens

Each tool call shows up as one AssistantMessage, and the result message closes the run. Here is the same loop in both SDKs:

import { query } from "@anthropic-ai/claude-agent-sdk";

for await (const message of query({
  prompt: "Fix the failing tests in auth.ts",
  options: { allowedTools: ["Read", "Edit", "Bash(npm test *)"], maxTurns: 30 },
})) {
  if (message.type === "assistant") {
    for (const block of message.message.content) {
      if (block.type === "tool_use") console.log(`Using ${block.name}`);
      else if (block.type === "text") console.log(block.text);
    }
  } else if (message.type === "result") {
    console.log(`${message.subtype} after ${message.num_turns} turns, $${message.total_cost_usd}`);
  }
}

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ResultMessage, TextBlock, ToolUseBlock


async def main():
    options = ClaudeAgentOptions(allowed_tools=["Read", "Edit", "Bash(npm test *)"], max_turns=30)
    async for message in query(prompt="Fix the failing tests in auth.ts", options=options):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    print(f"Using {block.name}")
                elif isinstance(block, TextBlock):
                    print(block.text)
        elif isinstance(message, ResultMessage):
            print(f"{message.subtype} after {message.num_turns} turns, ${message.total_cost_usd}")


asyncio.run(main())

A single-shot query() raises after it yields an error result, such as hitting max turns, so production code wraps the loop in try. The SDK docs' examples do the same.

Stream text token by token

Partial messages give you text as Claude writes it. Set includePartialMessages (TypeScript) or include_partial_messages (Python) to true, then look for content_block_delta events whose delta type is text_delta:

for await (const message of query({ prompt, options: { includePartialMessages: true } })) {
  if (message.type === "stream_event" && message.event.type === "content_block_delta" && message.event.delta.type === "text_delta") {
    process.stdout.write(message.event.delta.text);
  }
}

from claude_agent_sdk import query, ClaudeAgentOptions
from claude_agent_sdk.types import StreamEvent

async for message in query(prompt=prompt, options=ClaudeAgentOptions(include_partial_messages=True)):
    if isinstance(message, StreamEvent):
        event = message.event
        if event.get("type") == "content_block_delta" and event.get("delta", {}).get("type") == "text_delta":
            print(event["delta"].get("text", ""), end="", flush=True)

Stream events come from the main session only. Subagents don't forward token deltas, so attribute subagent output from complete messages, which carry parent_tool_use_id. A content_block_start event with a tool_use block tells you a tool call is starting before its input has finished streaming.

Heartbeats and subagent summaries

TypeScript has two more progress signals. While a tool call runs in the main conversation, a tool_progress message with heartbeat: true arrives every 30 seconds with the tool name and elapsed seconds, so you can tell a long call from a stalled session. Heartbeats need Agent SDK v0.3.214 or later and aren't sent for tool calls inside a subagent. For subagents and background tasks, system messages with subtype task_progress report a description, token usage and the last tool used; set agentProgressSummaries: true and their summary field carries a one-line summary of what the subagent is doing.

for await (const message of query({ prompt, options: { agentProgressSummaries: true } })) {
  if (message.type === "tool_progress" && message.heartbeat) {
    console.log(`${message.tool_name} still running after ${message.elapsed_time_seconds}s`);
  } else if (message.type === "system" && message.subtype === "task_progress") {
    console.log(`Subagent: ${message.summary ?? message.description}`);
  }
}

The Python reference lists TaskStartedMessage, TaskProgressMessage (description, usage, last_tool_name) and TaskNotificationMessage for background tasks, but no heartbeat message and no agentProgressSummaries option. In Python, the tool calls in AssistantMessage are the main progress signal.

On Claude Opus 4.7, Sonnet 4.6, Haiku 4.5 and earlier models, Claude also keeps a written todo list with the TaskCreate and TaskUpdate tools, which you can render from their tool_use blocks. Newer models don't get those tools unless you opt in, for example with CLAUDE_CODE_ENABLE_TODO_TOOLS=1 in the env option.

Seeing a run from outside the process

Messages only reach the process that called query(). When the agent runs in a container, a queue worker or a cron job, nobody is watching that process, so you'd need your own channel to a screen and a way to answer when the agent needs a decision. Taskhold provides both: a live board per person that shows each task, its steps and its latest progress message, and a Needs you list where you answer blocked tasks from any browser, including your phone.

Option 1: give the agent Taskhold's MCP server

The SDK connects to remote MCP servers over HTTP. Add Taskhold under mcpServers with type http and your API key in an Authorization header read from an environment variable, allow its tools with mcp__taskhold__*, and append the workflow to the system prompt:

import { query } from "@anthropic-ai/claude-agent-sdk";

const WORKFLOW = `You share a task list with the person through the taskhold tools. They watch a live board.
- Start: taskhold_begin with title, project "web-app", a short list of steps and agent "Upgrade agent (SDK)".
- Each step done: taskhold_complete on the step with start_next: true.
- Long step: taskhold_progress on the step with current, total and a short message. Every few minutes at most.
- Need a decision: taskhold_block with one question (2 to 4 options if it's a choice), then
  taskhold_updates with task_id, since and wait: 50 until the reply arrives.
- Done: taskhold_complete on the task with what changed and how you checked it.`;

for await (const message of query({
  prompt: "Upgrade date-fns to the latest major version and fix what breaks",
  options: {
    mcpServers: {
      taskhold: {
        type: "http",
        url: "https://taskhold.com/api/mcp",
        headers: { Authorization: `Bearer ${process.env.TASKHOLD_API_KEY}` },
      },
    },
    allowedTools: ["mcp__taskhold__*", "Read", "Edit", "Bash(npm test *)"],
    systemPrompt: { type: "preset", preset: "claude_code", append: WORKFLOW },
    maxTurns: 80,
    maxBudgetUsd: 5,
  },
})) {
  if (message.type === "system" && message.subtype === "init") console.log("MCP servers:", message.mcp_servers);
  if (message.type === "result") console.log(message.subtype);
}

import asyncio
import os
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage, SystemMessage

WORKFLOW = """..."""  # the same instructions as the TypeScript example


async def main():
    options = ClaudeAgentOptions(
        mcp_servers={
            "taskhold": {
                "type": "http",
                "url": "https://taskhold.com/api/mcp",
                "headers": {"Authorization": f"Bearer {os.environ['TASKHOLD_API_KEY']}"},
            }
        },
        allowed_tools=["mcp__taskhold__*", "Read", "Edit", "Bash(npm test *)"],
        system_prompt={"type": "preset", "preset": "claude_code", "append": WORKFLOW},
        max_turns=80,
        max_budget_usd=5,
    )
    async for message in query(prompt="Upgrade date-fns to the latest major version and fix what breaks", options=options):
        if isinstance(message, SystemMessage) and message.subtype == "init":
            print("MCP servers:", message.data.get("mcp_servers"))
        elif isinstance(message, ResultMessage):
            print(message.subtype)


asyncio.run(main())

• Use type http, not streamable-http. JSON config files accept streamable-http as an alias, but the SDKs' McpHttpServerConfig type declares only http for servers you pass in code.
• Allow the tools by name. MCP tools need permission before Claude can call them, and permissionMode acceptEdits doesn't approve them; the mcp__taskhold__* wildcard approves this one server and nothing else.
• Pass the key in headers. The SDK doesn't open a browser or run an OAuth flow, so the sign-in that claude.ai and Claude Code use for Taskhold doesn't apply here. Create a key in Taskhold Settings and give each agent its own, so you can revoke one without touching the others.
• Check the init message. When a server can't connect, its status there is failed or needs-auth and the run continues without its tools.

The appended text works even when you set settingSources: [] to keep host files out of the run. With default options the SDK also loads CLAUDE.md, so if the repo already has the CLAUDE.md task workflow you can leave the append out.

What you see while it runs

• A card per task an agent is working on: which agent, how long it has run, an estimate of time left once there is enough to go on, and the message from its last progress report.
• A progress bar split into one segment per step, so you see which step is running and which one is blocked.
• Blocked tasks at the top under Needs you, with the agent's question, one-tap options when it offers them, and Send and unblock.
• A status dot: Live if the agent reported in the last two minutes, Quiet after that, and No word, with how long ago, after fifteen minutes. A stalled agent stands out without you reading its terminal.
• Optional desktop notifications when an agent gets blocked or finishes, and a count in the tab title while you're in another tab.

When the agent calls taskhold_block, the task moves to Needs you with its question and any options as buttons. You answer from your phone, and the agent's taskhold_updates call with wait: 50 returns your reply as a note. Each wait is a tool call, so leave room in maxTurns for a few of them, or tell the agent to block and finish the run and let the next run read the answer. Answering a blocked coding agent covers how to word questions that get answered quickly.

Option 2: report progress from your code over REST

Some SDK apps shouldn't hand the agent more tools. Your code can report instead: create a task when the run starts, post a progress message as tool calls come through, and mark it done or blocked from the result. The same API key works for REST and MCP:

import { query } from "@anthropic-ai/claude-agent-sdk";

const TH = "https://taskhold.com/api";
const headers = {
  Authorization: `Bearer ${process.env.TASKHOLD_API_KEY}`,
  "X-Taskhold-Agent": "Upgrade agent (SDK)", // the name on the board
  "Content-Type": "application/json",
};
const th = (method: string, path: string, body: unknown) =>
  fetch(`${TH}${path}`, { method, headers, body: JSON.stringify(body) });

const created = await th("POST", "/tasks", { title: "Upgrade date-fns", project: "web-app" });
const { task } = await created.json();
await th("PATCH", `/tasks/${task.id}`, { status: "in_progress" });

let calls = 0;
let lastReport = 0;
for await (const message of query({ prompt: "Upgrade date-fns and fix what breaks", options: { allowedTools: ["Read", "Edit", "Bash(npm test *)"] } })) {
  if (message.type === "assistant") {
    for (const block of message.message.content) {
      if (block.type !== "tool_use") continue;
      calls++;
      if (Date.now() - lastReport > 10_000) { // one report every 10 s is plenty
        lastReport = Date.now();
        await th("POST", `/tasks/${task.id}/progress`, { message: `Tool call ${calls}: ${block.name}` }).catch(() => {});
      }
    }
  } else if (message.type === "result") {
    await th("PATCH", `/tasks/${task.id}`, message.subtype === "success"
      ? { status: "done", completed_summary: message.result.slice(0, 5000) }
      : { status: "blocked", blocked_reason: `Run stopped (${message.subtype}). Rerun with a higher limit?` });
  }
}

• POST /api/tasks/{id}/progress takes percent (0 to 100), or current and total, and/or a message of up to 500 characters shown under the bar. It works on a step or on a task without steps.
• The API accepts 120 progress reports a minute and merges reports on one task that land within 5 seconds of each other, so a throttle like the one above keeps every report visible.
• For steps, POST /api/tasks/{id}/subtasks with {"steps": [...]} splits the bar into segments, and PATCHing a step's status to done fills one. The CI progress recipe shows the same calls with curl.
• To ask the person something, PATCH the task to blocked with blocked_reason and block_options (2 to 4 choices), then long-poll GET /api/events?task_id=<id>&since=<cursor>&wait=50 for their reply. Events covers the cursor.

The full schema is in /openapi.json. Both options can run together: the agent plans and reports steps through MCP while your code adds a heartbeat line from the stream.