Claude Code 源码解读之工具系统

qq_45925576

460人浏览 · 2026-04-17 11:35:29

qq_45925576 · 2026-04-17 11:35:29 发布

03 - Claude Code Tool System

一、概念解释

为什么 AI Agent 需要工具？

LLM 本身只能"说话"，无法实际操作计算机。工具系统让 Agent 获得：

文件操作 — 读取、编辑、创建文件
代码搜索 — 按文件名模式搜索、按内容搜索
命令执行 — 运行 shell 命令
网络访问 — 搜索网页、抓取URL内容
子代理 — 启动子 Agent 处理子任务

工具系统的设计原则

声明式定义 — 每个工具用 Zod schema 声明输入输出
权限控制 — 工具调用前需通过 checkPermissions 和 canUseTool 双重检查
并发安全 — 只读工具可并行（最大并发数默认 10），写操作串行
结果预算 — maxResultSizeChars 限制工具返回结果大小，防止撑爆上下文

二、工具分类表

注意：工具文件名（如 FileReadTool）和工具实际 name（如 Read）不同。
API 使用的是 name 属性，代码引用使用文件名/变量名。

分类	文件名 / 变量名	实际 name	只读	并发安全
文件读取	FileReadTool	`Read`	是	是
文件编辑	FileEditTool	`Edit`	否	否
文件写入	FileWriteTool	`Write`	否	否
笔记本	NotebookEditTool	`NotebookEdit`	否	否
搜索	GlobTool	`Glob`	是	是
搜索	GrepTool	`Grep`	是	是
命令执行	BashTool	`Bash`	否	否
命令执行	PowerShellTool	`PowerShell`	否	否
网络	WebFetchTool	`WebFetch`	是	是
网络	WebSearchTool	`WebSearch`	是	是
代理	AgentTool	`Agent`	否	否
任务输出	TaskOutputTool	`TaskOutput`	是	是
任务停止	TaskStopTool	`TaskStop`	否	否
任务管理	TaskCreateTool 等	`TaskCreate` 等	否	否
计划	EnterPlanModeTool	`EnterPlanMode`	是	是
计划	ExitPlanModeV2Tool	`ExitPlanMode`	是	是
交互	AskUserQuestionTool	`AskUserQuestion`	是	是
Skill	SkillTool	`Skill`	是	是
MCP	动态生成	`mcp__server__tool`	看情况	看情况
调度	CronCreateTool 等	`CronCreate` 等	否	否
团队	TeamCreateTool 等	`TeamCreate` 等	否	否
工作树	EnterWorktreeTool 等	`EnterWorktree` 等	否	否
搜索	ToolSearchTool	`ToolSearch`	是	是

三、核心流程图

四、解决什么问题？

工具系统解决的核心问题：为 LLM 提供标准化的"手脚"，使其能够实际操作计算机完成编程任务。

关键设计决策：

工具定义统一接口（Tool 类型），每个工具有标准化的 name、schema、call 方法
工具注册中心（assembleToolPool()）统一管理内置工具 + MCP 工具的可用性
工具编排器（runTools()）通过 partitionToolCalls 智能调度并行/串行执行

五、核心代码详解

5.1 工具注册中心

// src/tools.ts — 工具获取的三个层次

// 第一层：所有基础工具（无过滤）
export function getAllBaseTools(): Tools {
  return [
    AgentTool, TaskOutputTool, BashTool,
    ...(hasEmbeddedSearchTools() ? [] : [GlobTool, GrepTool]),
    ExitPlanModeV2Tool, FileReadTool, FileEditTool, FileWriteTool,
    NotebookEditTool, WebFetchTool, TodoWriteTool, WebSearchTool,
    TaskStopTool, AskUserQuestionTool, SkillTool, EnterPlanModeTool,
    // ... 条件性工具（根据 feature flag 动态添加）：
    // ConfigTool (ant-only), TungstenTool (ant-only),
    // TaskCreateTool/Get/Update/List (todo v2),
    // EnterWorktreeTool/ExitWorktreeTool,
    // SendMessageTool, ListPeersTool, TeamCreateTool, TeamDeleteTool,
    // WorkflowTool, SleepTool, CronCreate/Delete/List,
    // RemoteTriggerTool, MonitorTool, BriefTool,
    // PowerShellTool, SnipTool, ToolSearchTool,
    ListMcpResourcesTool, ReadMcpResourceTool,
  ]
}

// 第二层：权限过滤
export const getTools = (permissionContext: ToolPermissionContext): Tools => {
  // --bare 简单模式：只有 Bash/Read/Edit
  if (isEnvTruthy(process.env.CLAUDE_CODE_SIMPLE)) {
    return filterToolsByDenyRules([BashTool, FileReadTool, FileEditTool], permissionContext)
  }
  // 正常模式：获取所有工具 → isEnabled → denyRules
  const tools = getAllBaseTools().filter(tool => !specialTools.has(tool.name))
  let allowedTools = filterToolsByDenyRules(tools, permissionContext)
  const isEnabled = allowedTools.map(_ => _.isEnabled())
  return allowedTools.filter((_, i) => isEnabled[i])
}

// 第三层：合并 MCP 工具（内置工具排序 + MCP 工具排序 + 去重）
export function assembleToolPool(
  permissionContext: ToolPermissionContext,
  mcpTools: Tools,
): Tools {
  const builtInTools = getTools(permissionContext)
  const allowedMcpTools = filterToolsByDenyRules(mcpTools, permissionContext)
  // 按名称排序内置工具和 MCP 工具，内置工具优先（同名冲突时内置胜出）
  const byName = (a: Tool, b: Tool) => a.name.localeCompare(b.name)
  return uniqBy(
    [...builtInTools].sort(byName).concat(allowedMcpTools.sort(byName)),
    'name',
  )
}

// 简化版合并（不排序，用于 token 计数等场景）
export function getMergedTools(
  permissionContext: ToolPermissionContext,
  mcpTools: Tools,
): Tools {
  return [...getTools(permissionContext), ...mcpTools]
}

5.2 工具接口定义

// src/Tool.ts — Tool 接口核心

// ★ buildTool 是工具构建的核心函数
// 它接受一个 ToolDef（部分定义），自动填充默认实现，返回完整的 Tool 对象
// TypeScript 推断确保所有必填字段都已提供
export function buildTool<D>(def: D): BuiltTool<D>

// 完整的 Tool 接口（关键字段）
export type Tool<I = any, O = any> = {
  // ===== 标识 =====
  readonly name: string                // 工具唯一名称（如 'Read', 'Edit'）
  aliases?: string[]                   // 向后兼容的旧名称
  searchHint?: string                  // ToolSearch 用的 3-10 词描述
  readonly shouldDefer?: boolean       // 是否延迟加载（减少 token 消耗）
  readonly alwaysLoad?: boolean        // 是否始终加载（跳过延迟）

  // ===== Schema =====
  readonly inputSchema: I              // Zod 输入 schema
  readonly inputJSONSchema?: ToolInputJSONSchema  // MCP 工具的 JSON schema
  outputSchema?: z.ZodType<unknown>    // 可选的输出 schema
  readonly strict?: boolean            // 严格模式标志

  // ===== 可用性判断 =====
  isEnabled(): boolean                 // 工具是否可用（feature flag / 配置检查）
  isConcurrencySafe(input: I): boolean // 此输入是否可与其他工具并发执行
  isReadOnly(input: I): boolean        // 此输入是否只读（不修改文件系统）
  isDestructive?(input: I): boolean    // 此输入是否不可逆（如 rm）
  interruptBehavior?(): 'cancel' | 'block'  // 中断行为
  isOpenWorld?(input: I): boolean      // 是否开放式工具（如 WebSearch）
  requiresUserInteraction?(): boolean  // 是否需要用户交互
  isMcp?: boolean                      // 是否为 MCP 工具
  isLsp?: boolean                      // 是否为 LSP 工具

  // ===== 描述与提示 =====
  description(input?: any, options?: any): Promise<string>  // 工具描述
  prompt(options?: any): Promise<string>                     // 系统提示中的用法说明
  userFacingName(input?: I): string        // UI 显示名称
  userFacingNameBackgroundColor?(input?: I): string
  toAutoClassifierInput(input: I): string  // 自动分类器输入

  // ===== 权限与验证 =====
  checkPermissions(input: I, context: ToolUseContext): Promise<PermissionResult>
  validateInput?(input: I, context: ToolUseContext): Promise<ValidationResult>

  // ===== 核心执行 =====
  call(
    input: I,
    context: ToolUseContext,
    canUseTool: CanUseToolFn,
    parentMessage?: AssistantMessage,
    onProgress?: (progress: ToolProgress) => void,
  ): Promise<ToolResult<O>>

  // ===== 结果映射 =====
  maxResultSizeChars: number  // 结果最大字符数（超出则持久化到磁盘）
  mapToolResultToToolResultBlockParam(content: O, toolUseID: string): ToolResultBlockParam

  // ===== UI 渲染 =====
  renderToolUseMessage?(input: I, options): ReactNode
  renderToolResultMessage?(content: O, progressMessages, options): ReactNode
  renderToolUseProgressMessage?(progressMessages, options): ReactNode
  renderToolUseRejectedMessage?(input: I, options): ReactNode
  renderToolUseErrorMessage?(result, options): ReactNode
  renderGroupedToolUse?(toolUses, options): ReactNode
  // ... 更多渲染方法
}

// ===== ToolUseContext — 工具执行环境 =====
export type ToolUseContext = {
  options: {
    commands: Command[]                     // 可用命令列表
    tools: Tools                            // 可用工具列表
    mainLoopModel: string                   // 当前模型
    thinkingConfig: ThinkingConfig          // 思考配置
    mcpClients: MCPServerConnection[]       // MCP 客户端连接
    agentDefinitions: AgentDefinitionsResult // 代理定义
    refreshTools?: () => Tools              // 刷新工具列表
  }
  abortController: AbortController          // 中止控制器
  messages: Message[]                       // 当前消息列表
  agentId?: AgentId                         // 子代理 ID（undefined = 主线程）
  getAppState: () => AppState               // 读取应用状态
  setAppState: SetAppState                  // 更新应用状态（子代理可能降级为 no-op）
  setAppStateForTasks?: SetAppState         // 后台任务专用的状态更新（始终有效）
  readFileState: FileStateCache             // 文件读取缓存（去重用）
  contentReplacementState?: ContentReplacementState  // 内容替换状态
  queryTracking?: { chainId: string; depth: number }  // 查询链追踪
  getRenderedSystemPrompt?: () => string    // 获取已渲染的系统提示（fork 子代理用）
  addNotification?: (notification) => void  // 添加通知
  pushApiMetricsEntry?: (ttftMs: number) => void  // API 指标推送
}

5.3 以 EnterPlanModeTool 为例的工具实现

// src/tools/EnterPlanModeTool/EnterPlanModeTool.ts
export const EnterPlanModeTool: Tool = buildTool({
  name: 'EnterPlanMode',
  searchHint: 'enter plan-only mode for complex tasks',
  shouldDefer: true,  // 延迟加载，减少 token 消耗

  async description() {
    return 'Requests permission to enter plan mode for complex tasks'
  },

  async prompt(options) {
    return getEnterPlanModeToolPrompt(options)
  },

  inputSchema: lazySchema(() => z.strictObject({})),  // 无参数

  isConcurrencySafe() { return true },  // 只读，可并发
  isReadOnly() { return true },

  async call(_input, context) {
    // 安全检查：子代理不能使用
    if (context.agentId) {
      throw new Error('Cannot use in agent contexts')
    }

    // 切换到 plan 模式 — 限制为只读工具
    context.setAppState(prev => ({
      ...prev,
      toolPermissionContext: applyPermissionUpdate(
        prepareContextForPlanMode(prev.toolPermissionContext),
        { type: 'setMode', mode: 'plan', destination: 'session' },
      ),
    }))

    return { data: { message: 'Entered plan mode.' } }
  },

  mapToolResultToToolResultBlockParam({ message }, toolUseID) {
    return {
      type: 'tool_result',
      content: `${message}\n\nIn plan mode, you should:\n1. Explore codebase\n2. Design approach\n3. Use ExitPlanMode when ready`,
      tool_use_id: toolUseID,
    }
  },
})

5.4 工具执行与权限检查

// src/services/tools/toolExecution.ts
// 工具执行是一个 async generator（不是普通 async 函数）
// 因为需要流式 yield 进度消息

export async function* runToolUse(
  toolUse: ToolUseBlock,
  assistantMessage: AssistantMessage,
  canUseTool: CanUseToolFn,
  toolUseContext: ToolUseContext,
): AsyncGenerator<MessageUpdateLazy, void> {
  // ① 查找工具（支持别名回退）
  let tool = findToolByName(toolUseContext.options.tools, toolUse.name)
  if (!tool) {
    // 检查是否为已弃用的别名
    tool = findToolByAlias(toolUseContext.options.tools, toolUse.name)
  }
  if (!tool) {
    yield { message: createToolNotFoundMessage(toolUse) }
    return
  }

  // ② 执行完整的权限和调用流程（内部函数 checkPermissionsAndCallTool）
  // 流程：
  //   a. Zod 解析输入 → tool.inputSchema.safeParse(input)
  //   b. 工具自定义验证 → tool.validateInput(input, context)
  //   c. PreToolUse hooks 执行 → 执行用户配置的 hooks
  //   d. 权限检查 → canUseTool(tool, input, context)
  //      - 返回 'allow' → 继续
  //      - 返回 'deny' → 返回拒绝消息
  //      - 返回 'ask' → 弹出用户确认对话框
  //   e. 执行工具 → tool.call(input, context, canUseTool, assistantMessage, onProgress)
  //   f. PostToolUse hooks 执行
  //   g. 结果映射 → tool.mapToolResultToToolResultBlockParam(result, toolUse.id)
  //   h. Yield 消息更新
}

5.5 工具并发编排

// src/services/tools/toolOrchestration.ts

// 并发上限（可通过环境变量覆盖）
function getMaxToolUseConcurrency(): number {
  return parseInt(
    process.env.CLAUDE_CODE_MAX_TOOL_USE_CONCURRENCY || '', 10
  ) || 10  // 默认最大 10 并发
}

export async function* runTools(
  toolUseMessages: ToolUseBlock[],
  assistantMessages: AssistantMessage[],
  canUseTool: CanUseToolFn,
  toolUseContext: ToolUseContext,
): AsyncGenerator<MessageUpdate, void> {
  let currentContext = toolUseContext

  // ★ 分区算法：将工具调用分为连续的可并发批次和不可并发批次
  for (const { isConcurrencySafe, blocks } of partitionToolCalls(
    toolUseMessages, toolUseContext
  )) {
    if (isConcurrencySafe) {
      // 并行执行：使用 all() 并发运行（受 maxConcurrency 限制）
      for await (const update of runToolsConcurrently(
        blocks, assistantMessages, canUseTool, currentContext,
        getMaxToolUseConcurrency(),
      )) {
        yield { message: update.message, newContext: currentContext }
      }
    } else {
      // 串行执行：依次运行，每次更新 context
      for await (const update of runToolsSerially(
        blocks, assistantMessages, canUseTool, currentContext,
      )) {
        currentContext = update.newContext ?? currentContext
        yield { message: update.message, newContext: currentContext }
      }
    }
  }
}

// 分区逻辑：连续的 concurrency-safe 工具合并为一个批次
// 不 safe 的工具各自成为独立批次（不能与其他工具并发）
function partitionToolCalls(toolUseMessages, toolUseContext): Batch[] {
  return toolUseMessages.reduce((acc, toolUse) => {
    const tool = findToolByName(toolUseContext.options.tools, toolUse.name)
    const parsedInput = tool?.inputSchema.safeParse(toolUse.input)
    const isConcurrencySafe = parsedInput?.success
      ? Boolean(tool?.isConcurrencySafe(parsedInput.data))
      : false  // 解析失败则保守地认为不安全

    // 如果上一个批次也是 safe 的，合并到同一批次
    if (isConcurrencySafe && acc[acc.length - 1]?.isConcurrencySafe) {
      acc[acc.length - 1].blocks.push(toolUse)
    } else {
      // 否则新建一个批次
      acc.push({ isConcurrencySafe, blocks: [toolUse] })
    }
    return acc
  }, [])
}

// 示例：LLM 请求 [Read, Glob, Edit, Grep, Write]
// 分区结果：
//   Batch 1: safe=true  → [Read, Glob]        并行执行
//   Batch 2: safe=false → [Edit]              串行执行
//   Batch 3: safe=true  → [Grep]              并行执行（只有一个，但仍是 safe 批次）
//   Batch 4: safe=false → [Write]             串行执行

六、示例代码：自定义工具

// custom-tool-example.ts — 演示如何使用 buildTool 创建工具
import { z } from 'zod'
import { buildTool, type ToolDef } from './Tool'

// ① 定义输入输出 Schema
const inputSchema = z.strictObject({
  directory: z.string().describe('要列出的目录路径'),
  pattern: z.string().optional().describe('文件名过滤模式'),
})

const outputSchema = z.object({
  files: z.array(z.string()).describe('找到的文件列表'),
  count: z.number().describe('文件数量'),
})

// ② 使用 buildTool 构建工具
// ToolDef 是部分定义类型 — 只需提供必填字段，buildTool 自动填充默认值
const ListFilesTool: ToolDef<typeof inputSchema, typeof outputSchema> = buildTool({
  name: 'list_files',
  searchHint: 'list files in a directory',

  async description() {
    return 'Lists all files in the specified directory, optionally filtered by pattern'
  },

  async prompt() {
    return `Use this tool to list files in a directory.`
  },

  get inputSchema() { return inputSchema },
  get outputSchema() { return outputSchema },

  isConcurrencySafe() { return true },   // 只读，可并发
  isReadOnly() { return true },
  maxResultSizeChars: 50_000,            // 结果最大 50K 字符

  // checkPermissions 默认返回 allow — 可选覆盖
  async checkPermissions(input, context) {
    return { behavior: 'allow', updatedInput: input }
  },

  async call(input, context) {
    // 实际的文件列表逻辑
    const files = ['src/index.ts', 'src/utils.ts', 'README.md']
    const filtered = input.pattern
      ? files.filter(f => f.includes(input.pattern!))
      : files

    return {
      data: {
        files: filtered,
        count: filtered.length,
      }
    }
  },

  mapToolResultToToolResultBlockParam(result, toolUseID) {
    return {
      type: 'tool_result',
      content: `Found ${result.count} files:\n${result.files.map(f => `- ${f}`).join('\n')}`,
      tool_use_id: toolUseID,
    }
  },
})

export { ListFilesTool }

七、设计要点总结

统一接口 — 所有工具通过 buildTool() 构建实现相同的 Tool 接口，Agent Loop 无需关心具体工具细节
声明式 Schema — Zod schema 同时用于验证和生成 LLM 看到的工具描述（inputSchema.safeParse 用于验证 + inputJSONSchema 用于 API 展示）
权限分层 — checkPermissions（工具级）+ canUseTool（全局级）+ denyRules（注册时过滤），三层权限控制
并发优化 — isConcurrencySafe() 标记 + partitionToolCalls 分区，让只读工具并行执行，默认最大 10 并发
结果预算 — maxResultSizeChars 限制返回大小，超出时自动持久化到磁盘并通过 TaskOutputTool 按需读取
条件注册 — feature flag、环境变量、isEnabled() 三重门控控制工具是否出现在列表中，减少不必要的 token 消耗
延迟加载 — shouldDefer: true 让工具延迟到 ToolSearch 搜索时才加载详情，节省上下文空间