Agent SDK Advanced

If you learned the basics of query() in Level 4, here we cover the advanced patterns needed for production-grade agent systems.

Streaming

Show long-running work to the user in real time:

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

async function streamingQuery(prompt: string) {
  const messages: SDKMessage[] = [];

  for await (const message of query({
    prompt,
    options: { maxTurns: 10 }
  })) {
    messages.push(message);

    // Real-time output
    if (message.type === "assistant") {
      for (const block of message.content) {
        if (block.type === "text") {
          process.stdout.write(block.text);
        }
      }
    }

    // Track tool usage
    if (message.type === "tool_use") {
      console.log(`\n🔧 Tool used: ${message.name}`);
    }
  }

  return messages;
}

Advanced Error Handling

Retry logic

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

interface RetryOptions {
  maxRetries: number;
  backoffMs: number;
  onRetry?: (attempt: number, error: Error) => void;
}

async function queryWithRetry(
  prompt: string,
  retryOptions: RetryOptions = { maxRetries: 3, backoffMs: 1000 }
): Promise<any[]> {
  let lastError: Error;

  for (let attempt = 1; attempt <= retryOptions.maxRetries; attempt++) {
    try {
      const messages: any[] = [];
      for await (const msg of query({ prompt, options: { maxTurns: 15 } })) {
        messages.push(msg);
      }
      return messages;
    } catch (error) {
      lastError = error as Error;
      retryOptions.onRetry?.(attempt, lastError);

      if (attempt < retryOptions.maxRetries) {
        const delay = retryOptions.backoffMs * Math.pow(2, attempt - 1);
        console.log(`⏳ Retrying in ${delay}ms (${attempt}/${retryOptions.maxRetries})...`);
        await new Promise(r => setTimeout(r, delay));
      }
    }
  }

  throw new Error(`Failed after ${retryOptions.maxRetries} attempts: ${lastError!.message}`);
}

Timeout handling

async function queryWithTimeout(
  prompt: string,
  timeoutMs: number = 60000
): Promise<any[]> {
  const timeoutPromise = new Promise<never>((_, reject) => {
    setTimeout(() => reject(new Error(`Timeout: exceeded ${timeoutMs}ms`)), timeoutMs);
  });

  const queryPromise = (async () => {
    const messages: any[] = [];
    for await (const msg of query({ prompt, options: { maxTurns: 20 } })) {
      messages.push(msg);
    }
    return messages;
  })();

  return Promise.race([queryPromise, timeoutPromise]);
}

State Management Patterns

Agent session management

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

interface AgentSession {
  id: string;
  history: Array<{ prompt: string; result: string; timestamp: Date }>;
  context: Record<string, any>;
}

class AgentSessionManager {
  private sessions = new Map<string, AgentSession>();

  createSession(id: string): AgentSession {
    const session: AgentSession = {
      id,
      history: [],
      context: {}
    };
    this.sessions.set(id, session);
    return session;
  }

  async runWithSession(sessionId: string, prompt: string): Promise<string> {
    const session = this.sessions.get(sessionId);
    if (!session) throw new Error(`No such session: ${sessionId}`);

    // Include previous context
    const contextSummary = session.history.length > 0
      ? `Summary of previous work:\n${session.history
          .slice(-3) // only the most recent 3
          .map(h => `- ${h.prompt}: ${h.result.slice(0, 100)}...`)
          .join("\n")}\n\n`
      : "";

    const fullPrompt = contextSummary + prompt;

    const messages: any[] = [];
    for await (const msg of query({ prompt: fullPrompt, options: { maxTurns: 15 } })) {
      messages.push(msg);
    }

    const result = extractText(messages);

    // Update history
    session.history.push({
      prompt,
      result,
      timestamp: new Date()
    });

    return result;
  }

  getSession(id: string): AgentSession | undefined {
    return this.sessions.get(id);
  }
}

// Usage
const manager = new AgentSessionManager();
const session = manager.createSession("project-alpha");
await manager.runWithSession("project-alpha", "Analyze the structure of the src/ folder");
await manager.runWithSession("project-alpha", "Fix the issues you found");

Dynamic Tool Control

Dynamically change the allowed tools per phase of work:

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

const TOOL_PRESETS = {
  readOnly: ["Read", "Glob", "Grep"],
  analyze: ["Read", "Glob", "Grep", "Bash"],
  write: ["Read", "Edit", "Write", "Bash"],
  full: ["Read", "Edit", "Write", "Bash", "Glob", "Grep"]
} as const;

async function phaseBasedExecution(codebase: string) {
  // Phase 1: analyze in read-only mode
  console.log("Phase 1: analyzing the codebase (read-only)");
  const analysisMessages: any[] = [];
  for await (const msg of query({
    prompt: `Analyze ${codebase} and output a JSON list of files that need improvement`,
    options: {
      allowedTools: [...TOOL_PRESETS.readOnly],
      maxTurns: 10
    }
  })) {
    analysisMessages.push(msg);
  }

  // Phase 2: make actual edits
  console.log("Phase 2: editing code (writes allowed)");
  const fixMessages: any[] = [];
  for await (const msg of query({
    prompt: `Based on the following analysis results, proceed with the fixes:\n${extractText(analysisMessages)}`,
    options: {
      allowedTools: [...TOOL_PRESETS.write],
      maxTurns: 30
    }
  })) {
    fixMessages.push(msg);
  }

  return extractText(fixMessages);
}

Metric Collection

Measure agent performance:

interface AgentMetrics {
  totalTurns: number;
  toolsUsed: Record<string, number>;
  durationMs: number;
  inputTokens: number;
  outputTokens: number;
}

async function queryWithMetrics(
  prompt: string
): Promise<{ result: string; metrics: AgentMetrics }> {
  const startTime = Date.now();
  const metrics: AgentMetrics = {
    totalTurns: 0,
    toolsUsed: {},
    durationMs: 0,
    inputTokens: 0,
    outputTokens: 0
  };
  const messages: any[] = [];

  for await (const msg of query({ prompt, options: { maxTurns: 20 } })) {
    messages.push(msg);

    if (msg.type === "assistant") {
      metrics.totalTurns++;
    }
    if (msg.type === "tool_use") {
      metrics.toolsUsed[msg.name] = (metrics.toolsUsed[msg.name] || 0) + 1;
    }
    // Token usage (when a usage field is present)
    if (msg.usage) {
      metrics.inputTokens += msg.usage.input_tokens || 0;
      metrics.outputTokens += msg.usage.output_tokens || 0;
    }
  }

  metrics.durationMs = Date.now() - startTime;

  return {
    result: extractText(messages),
    metrics
  };
}

// Usage
const { result, metrics } = await queryWithMetrics("Review all of the code in src/");
console.log(`Done: ${metrics.durationMs}ms, ${metrics.totalTurns} turns, tools: ${JSON.stringify(metrics.toolsUsed)}`);

Queue-Based Task Management

Process bulk work through a queue:

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

class AgentQueue {
  private queue: Array<{ prompt: string; resolve: Function; reject: Function }> = [];
  private running = 0;
  private maxConcurrent: number;

  constructor(maxConcurrent = 3) {
    this.maxConcurrent = maxConcurrent;
  }

  async add(prompt: string): Promise<string> {
    return new Promise((resolve, reject) => {
      this.queue.push({ prompt, resolve, reject });
      this.process();
    });
  }

  private async process() {
    if (this.running >= this.maxConcurrent || this.queue.length === 0) return;

    const task = this.queue.shift()!;
    this.running++;

    try {
      const messages: any[] = [];
      for await (const msg of query({ prompt: task.prompt, options: { maxTurns: 15 } })) {
        messages.push(msg);
      }
      task.resolve(extractText(messages));
    } catch (err) {
      task.reject(err);
    } finally {
      this.running--;
      this.process(); // process the next task
    }
  }
}

// Usage: up to 3 running concurrently
const agentQueue = new AgentQueue(3);
const tasks = files.map(file =>
  agentQueue.add(`Find the bugs in the file ${file}`)
);
const results = await Promise.all(tasks);

Production checklist

• Apply retry logic to all errors
• Prevent infinite waits with timeouts
• Track cost with metric collection
• Limit concurrency with a queue (to guard against API rate limits)
• Include the session ID in logs (for traceability)