Claude Agent SDK Todo 任务追踪指南

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

try {
  for await (const message of query({
    prompt: "Optimize my React app performance and track progress with todos",
    // Re-enable TodoWrite, which this example monitors. Without it, the SDK uses
    // Task tools instead and these tool_use blocks never appear. ENABLE_TODO_TOOLS
    // keeps the tools on models where Claude Code otherwise doesn't provide them.
    options: { maxTurns: 15, env: { ...process.env, CLAUDE_CODE_ENABLE_TASKS: "0", CLAUDE_CODE_ENABLE_TODO_TOOLS: "1" } }
  })) {
    // Todo updates are reflected in the message stream
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "tool_use" && block.name === "TodoWrite") {
          const todos = block.input.todos;

          console.log("Todo Status Update:");
          todos.forEach((todo, index) => {
            const status =
              todo.status === "completed" ? "✅" : todo.status === "in_progress" ? "🔧" : "❌";
            console.log(`${index + 1}. ${status} ${todo.content}`);
          });
        }
      }
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result,
  // such as when the maxTurns limit is hit.
  console.log(`Session ended with an error: ${error}`);
}

import asyncio

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock


async def main():
    try:
        async for message in query(
            prompt="Optimize my React app performance and track progress with todos",
            # Re-enable TodoWrite, which this example monitors. Without it, the SDK uses
            # Task tools instead and these tool_use blocks never appear. ENABLE_TODO_TOOLS
            # keeps the tools on models where Claude Code otherwise doesn't provide them.
            options=ClaudeAgentOptions(max_turns=15, env={"CLAUDE_CODE_ENABLE_TASKS": "0", "CLAUDE_CODE_ENABLE_TODO_TOOLS": "1"}),
        ):
            # Todo updates are reflected in the message stream
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, ToolUseBlock) and block.name == "TodoWrite":
                        todos = block.input["todos"]

                        print("Todo Status Update:")
                        for i, todo in enumerate(todos):
                            status = (
                                "✅"
                                if todo["status"] == "completed"
                                else "🔧"
                                if todo["status"] == "in_progress"
                                else "❌"
                            )
                            print(f"{i + 1}. {status} {todo['content']}")
    except Exception as error:
        # A single-shot query() raises after yielding an error result,
        # such as when the max_turns limit is hit.
        print(f"Session ended with an error: {error}")


asyncio.run(main())

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

class TodoTracker {
  private todos: any[] = [];

  displayProgress() {
    if (this.todos.length === 0) return;

    const completed = this.todos.filter((t) => t.status === "completed").length;
    const inProgress = this.todos.filter((t) => t.status === "in_progress").length;
    const total = this.todos.length;

    console.log(`\nProgress: ${completed}/${total} completed`);
    console.log(`Currently working on: ${inProgress} task(s)\n`);

    this.todos.forEach((todo, index) => {
      const icon =
        todo.status === "completed" ? "✅" : todo.status === "in_progress" ? "🔧" : "❌";
      const text = todo.status === "in_progress" ? todo.activeForm : todo.content;
      console.log(`${index + 1}. ${icon} ${text}`);
    });
  }

  async trackQuery(prompt: string) {
    try {
      for await (const message of query({
        prompt,
        // On every model, re-enable TodoWrite, which this tracker watches for.
        options: { maxTurns: 20, env: { ...process.env, CLAUDE_CODE_ENABLE_TASKS: "0", CLAUDE_CODE_ENABLE_TODO_TOOLS: "1" } }
      })) {
        if (message.type === "assistant") {
          for (const block of message.message.content) {
            if (block.type === "tool_use" && block.name === "TodoWrite") {
              this.todos = block.input.todos;
              this.displayProgress();
            }
          }
        }
      }
    } catch (error) {
      // A single-shot query() throws after yielding an error result,
      // such as when the maxTurns limit is hit.
      console.log(`Session ended with an error: ${error}`);
    }
  }
}

// Usage
const tracker = new TodoTracker();
await tracker.trackQuery("Build a complete authentication system with todos");

import asyncio

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock
from typing import List, Dict


class TodoTracker:
    def __init__(self):
        self.todos: List[Dict] = []

    def display_progress(self):
        if not self.todos:
            return

        completed = len([t for t in self.todos if t["status"] == "completed"])
        in_progress = len([t for t in self.todos if t["status"] == "in_progress"])
        total = len(self.todos)

        print(f"\nProgress: {completed}/{total} completed")
        print(f"Currently working on: {in_progress} task(s)\n")

        for i, todo in enumerate(self.todos):
            icon = (
                "✅"
                if todo["status"] == "completed"
                else "🔧"
                if todo["status"] == "in_progress"
                else "❌"
            )
            text = (
                todo["activeForm"]
                if todo["status"] == "in_progress"
                else todo["content"]
            )
            print(f"{i + 1}. {icon} {text}")

    async def track_query(self, prompt: str):
        try:
            async for message in query(
                prompt=prompt,
                # On every model, re-enable TodoWrite, which this tracker watches for.
                options=ClaudeAgentOptions(max_turns=20, env={"CLAUDE_CODE_ENABLE_TASKS": "0", "CLAUDE_CODE_ENABLE_TODO_TOOLS": "1"}),
            ):
                if isinstance(message, AssistantMessage):
                    for block in message.content:
                        if isinstance(block, ToolUseBlock) and block.name == "TodoWrite":
                            self.todos = block.input["todos"]
                            self.display_progress()
        except Exception as error:
            # A single-shot query() raises after yielding an error result,
            # such as when the max_turns limit is hit.
            print(f"Session ended with an error: {error}")


# Usage
async def main():
    tracker = TodoTracker()
    await tracker.track_query("Build a complete authentication system with todos")


asyncio.run(main())

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

try {
  for await (const message of query({
    prompt: "Optimize my React app performance and track progress with todos",
    // Keeps the Task tools on models where Claude Code otherwise doesn't provide them.
    options: { maxTurns: 15, env: { ...process.env, CLAUDE_CODE_ENABLE_TODO_TOOLS: "1" } },
  })) {
    if (message.type !== "assistant") continue;
    for (const block of message.message.content) {
      if (block.type !== "tool_use") continue;
      if (block.name === "TaskCreate") {
        const input = block.input as { subject: string };
        console.log(`+ ${input.subject}`);
      } else if (block.name === "TaskUpdate") {
        const input = block.input as {
          taskId?: string;
          id?: string;
          task_id?: string;
          status?: string;
        };
        const taskId = input.taskId ?? input.id ?? input.task_id;
        if (taskId && input.status) console.log(`  ${taskId} -> ${input.status}`);
      }
    }
  }
} catch (error) {
  // A single-shot query() throws after yielding an error result.
  console.log(`Session ended with an error: ${error}`);
}

import asyncio

from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, ToolUseBlock

async def main():
    try:
        async for message in query(
            prompt="Optimize my React app performance and track progress with todos",
            # Keeps the Task tools on models where Claude Code otherwise doesn't provide them.
            options=ClaudeAgentOptions(max_turns=15, env={"CLAUDE_CODE_ENABLE_TODO_TOOLS": "1"}),
        ):
            if not isinstance(message, AssistantMessage):
                continue
            for block in message.content:
                if not isinstance(block, ToolUseBlock):
                    continue
                if block.name == "TaskCreate":
                    print(f"+ {block.input['subject']}")
                elif block.name == "TaskUpdate" and block.input.get("status"):
                    task_id = (
                        block.input.get("taskId")
                        or block.input.get("id")
                        or block.input.get("task_id")
                    )
                    if task_id:
                        print(f"  {task_id} -> {block.input['status']}")
    except Exception as error:
        # A single-shot query() raises after yielding an error result.
        print(f"Session ended with an error: {error}")


asyncio.run(main())