Claude Code

Agent SDK Streaming 与插件

Agent SDK 的 streaming input、single message、插件加载、技能命名空间、session 控制、权限请求和 5m/1h 缓存影响。

Agent SDK 有两类容易被低估的运行时选择:输入模式和插件加载。输入模式决定 session 是否长期存在、能不能排队消息、上传图片、处理中断和恢复。插件加载决定 SDK 里的 agent 能否使用 Skills、agents、hooks 和 MCP server。

先读 Agent SDK API 参考速查和 Agent SDK 运行时模式,再用本页决定产品形态。

#输入模式对比

模式	适合	限制
Streaming input	长任务、聊天 UI、任务面板、图片输入、队列和中断	要处理 async generator 错误、权限暂停和取消
Single message	一次性分析、Lambda/Job、无交互脚本	不支持直接图片附件、动态排队、实时中断和自然多轮

官方推荐默认使用 streaming input。它把 agent 当成长生命周期进程,可以持续接收用户消息、执行工具、返回中间状态,并在同一 session 里保持上下文。

#Streaming input 结构

TypeScript streaming input

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

async function* messages(): 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: "Now focus on auth and payment paths"
    },
    parent_tool_use_id: null
  };
}

for await (const message of query({
  prompt: messages(),
  options: {
    maxTurns: 10,
    allowedTools: ["Read", "Grep"]
  }
})) {
  if (message.type === "result") console.log(message);
}

Python streaming input

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions
import asyncio

async def main():
    async def messages():
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Analyze this codebase for security issues",
            },
        }
        await asyncio.sleep(2)
        yield {
            "type": "user",
            "message": {
                "role": "user",
                "content": "Now focus on auth and payment paths",
            },
        }

    async with ClaudeSDKClient(ClaudeAgentOptions(max_turns=10)) as client:
        await client.query(messages())
        async for message in client.receive_response():
            print(message)

asyncio.run(main())

如果 TypeScript generator 抛错,SDK 可能把 session 结束表现成 aborted。Python generator 抛错可能只在 debug log 中出现并导致 session 挂住。产品里要把 generator 自己的文件读取、图片编码和网络请求包上错误处理。

#Single message 场景

One-shot query

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);
  }
}

如果结果是 error_max_turns 或其他 error result,单消息 query() 可能在 yield 最后 result 后再抛错。脚本要用 try/catch 包住整个 async loop,不要只检查成功分支。

#SDK 插件加载

SDK 只接受本地插件路径:

TypeScript plugins

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

for await (const message of query({
  prompt: "What custom capabilities do you have?",
  options: {
    plugins: [
      { type: "local", path: "./plugins/review" },
      { type: "local", path: "/opt/company/claude-plugins/security" }
    ],
    maxTurns: 3
  }
})) {
  if (message.type === "system" && message.subtype === "init") {
    console.log(message.plugins);
    console.log(message.skills);
    console.log(message.slash_commands);
  }
}

Python plugins

from claude_agent_sdk import query, ClaudeAgentOptions
import asyncio

async def main():
    async for message in query(
        prompt="What custom capabilities do you have?",
        options=ClaudeAgentOptions(
            plugins=[
                {"type": "local", "path": "./plugins/review"},
                {"type": "local", "path": "/opt/company/claude-plugins/security"},
            ],
            max_turns=3,
        ),
    ):
        print(message)

asyncio.run(main())

插件路径必须指向插件根目录,也就是包含 skills/、agents/、hooks/、commands/ 或 .claude-plugin/ 的目录。通过 CLI 安装过的插件也可以给 SDK 用,但要传入安装后的本地路径,通常在 ~/.claude/plugins/ 下。

#插件技能和命名空间

插件里的 Skills 会自动带插件名前缀,避免和内置命令冲突。

内容	SDK 中看到
插件 `review-kit` 的 skill `security`	`/review-kit:security`
插件旧 `commands/custom-command`	`/review-kit:custom-command`
插件 agent	出现在 system init 的 agents / capabilities 中
插件 MCP server	工具名通常变成 `mcp__server__tool`

如果你在产品 UI 里暴露这些能力,不要硬编码完整命令列表。读取 system/init 中的 skills、slash_commands、plugins 和可用 tools,再渲染给用户。

#权限和暂停

Streaming 模式下要处理三类暂停:

暂停	处理
工具审批	UI 展示工具名、参数摘要和风险,返回 allow/deny/defer
`AskUserQuestion`	用表单或选项让用户回答,再恢复 session
插件 hook 阻断	显示插件名、阻断理由和恢复建议

已经在 allowedTools、permission rule 或 permission mode 中自动批准的工具,不会再触发 canUseTool。必须每次都审的逻辑要放在 hook 或网关侧策略。

#缓存和成本

变化	5m / 1h 解释
Streaming session 连续输入	system、tools、settings 不变时最容易延续热缓存
Single message 无状态 Job	每个 Job 更像冷启动,5 分钟 TTL 复用有限
插件列表变化	skills、hooks、MCP 和 tool schema 变化会改变前缀
插件带 MCP server	如果 upfront 加载大量 tools,会显著增加 cache creation
使用 tool search	可减少工具定义进入前缀,改善大插件集的缓存稳定性
发送 `/compact`	会重写历史前缀,下一轮可能 cache miss