Claude Agent SDK 提供两种输入模式：流式输入（推荐）和单次消息输入。流式输入支持多轮对话、图片上传、工具调用、钩子和实时反馈，适合需要持续交互的复杂应用。单次消息输入更简单，适合一次性查询或 Lambda 等无状态环境，但不支持图片、队列、中断和钩子。选择时关键看是否需要持久会话上下文与交互能力。

Claude Agent SDK 输入模式：流式与单次消息对比

概览

Claude Agent SDK 支持两种不同的输入模式与 Agent 交互：

• 流式输入模式（默认，推荐） – 持久的交互式会话
• 单次消息输入 – 一次性查询，可使用会话状态和恢复

本文档解释两种模式的差异、优缺点和适用场景，帮助你在应用中做出合适的选择。

流式输入模式（推荐）

流式输入模式是使用 Claude Agent SDK 的首选方式。它提供 Agent 的全部能力，并支持丰富的交互体验。

该模式允许 Agent 作为一个长期运行的进程，接收用户输入、处理中断、展示权限请求，并管理会话状态。

工作原理

序列图略（保留原始 Mermaid 图，但这里用文字描述）
Your Application → Claude Agent：异步生成器初始化
消息1：发送文本 → Agent 执行工具 → 读写文件 → 流式返回部分响应 → 完成消息1
消息2 + 图片 → Agent 处理图片并执行 → 流式返回响应2
消息3入队 → 中断/取消 → 处理中断
会话保持存活，文件系统状态持续

（注：上游原文的 Mermaid 图已省略，建议在实际文档中保留原图。）

优势

• 图片上传 – 直接在消息中附加图片，用于视觉分析和理解
• 消息队列 – 发送多条消息顺序处理，支持中断
• 工具集成 – 会话期间完整访问所有工具和自定义 MCP 服务器
• 钩子支持 – 使用生命周期钩子在各个阶段自定义行为
• 实时反馈 – 看到正在生成的结果，而非仅最终结果
• 上下文持久化 – 在多次轮次中自然维持对话上下文

实现示例

import { query, type SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
import { readFile } from "fs/promises";

async function* generateMessages(): AsyncGenerator<SDKUserMessage> {
  // 第一条消息
  yield {
    type: "user",
    message: {
      role: "user",
      content: "Analyze this codebase for security issues"
    },
    parent_tool_use_id: null
  };

  // 等待条件或用户输入
  await new Promise((resolve) => setTimeout(resolve, 2000));

  // 后续带图片的消息
  yield {
    type: "user",
    message: {
      role: "user",
      content: [
        {
          type: "text",
          text: "Review this architecture diagram"
        },
        {
          type: "image",
          source: {
            type: "base64",
            media_type: "image/png",
            data: await readFile("diagram.png", "base64")
          }
        }
      ]
    },
    parent_tool_use_id: null
  };
}

// 处理流式响应
for await (const message of query({
  prompt: generateMessages(),
  options: {
    maxTurns: 10,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    TextBlock,
)
import asyncio
import base64


async def streaming_analysis():
    async def message_generator():
        # 第一条消息
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Analyze this codebase for security issues",
            },
        }

        # 等待条件
        await asyncio.sleep(2)

        # 后续带图片
        with open("diagram.png", "rb") as f:
            image_data = base64.b64encode(f.read()).decode()

        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": [
                    {"type": "text", "text": "Review this architecture diagram"},
                    {
                        "type": "image",
                        "source": {
                            "type": "base64",
                            "media_type": "image/png",
                            "data": image_data,
                        },
                    },
                ],
            },
        }

    # 使用 ClaudeSDKClient 进行流式输入
    options = ClaudeAgentOptions(max_turns=10, allowed_tools=["Read", "Grep"])

    async with ClaudeSDKClient(options) as client:
        # 发送流式输入
        await client.query(message_generator())

        # 处理响应
        async for message in client.receive_response():
            if isinstance(message, AssistantMessage):
                for block in message.content:
                    if isinstance(block, TextBlock):
                        print(block.text)


asyncio.run(streaming_analysis())

单次消息输入

单次消息输入更简单，但功能有限。

何时使用单次消息输入

在以下场景可以使用单次消息输入：

• 需要一次性的响应
• 不需要图片附件、钩子等功能
• 需要在无状态环境中运行，例如 Lambda 函数

限制

单次消息输入模式 不支持 以下功能：

• 直接在消息中附加图片
• 动态消息队列
• 实时中断
• 钩子集成
• 自然的连续多轮对话

实现示例

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

// 简单的一次性查询
for await (const message of query({
  prompt: "Explain the authentication flow",
  options: {
    maxTurns: 1,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

// 通过会话管理继续对话
for await (const message of query({
  prompt: "Now explain the authorization process",
  options: {
    continue: true,
    maxTurns: 1
  }
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.result);
  }
}

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage
import asyncio


async def single_message_example():
    # 简单的一次性查询，使用 query() 函数
    async for message in query(
        prompt="Explain the authentication flow",
        options=ClaudeAgentOptions(max_turns=1, allowed_tools=["Read", "Grep"]),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

    # 通过会话管理继续对话
    async for message in query(
        prompt="Now explain the authorization process",
        options=ClaudeAgentOptions(continue_conversation=True, max_turns=1),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)


asyncio.run(single_message_example())

常见问题

流式输入模式下如何中断正在执行的消息？

在生成器函数中，可以使用 return 或抛出异常来中断消息队列。SDK 会处理剩余的消息序列并终止当前会话。实际应用中，可以通过外部信号（如用户点击取消）控制生成器的 yield 逻辑。

单次消息输入能不能发送图片？

不能。单次消息输入模式不支持图片附件。如果需要发送图片，必须使用流式输入模式。

我可以在同一个会话中从单次消息切换到流式模式吗？

不能。模式的选择是在初始化 query() 调用时确定的。如果尝试混合使用，SDK 会抛出类型错误。必须在设计应用时就决定使用哪种输入模式。