DeepSeek Harness 系列（03）：工具系统——给 Agent 装上手

本文目录

Schema DSL 类型系统
阶段一：tools/pre-execute（执行前）
阶段二：tools/execute（环绕分派）
阶段三：tools/post-execute（执行后）
Guards：单调拒绝

为什么工具是 Agent 的核心 ----------------

LLM 本身只能输出文字。Agent 之所以能订机票、写代码、查数据库，靠的是工具——LLM 决定调什么、传什么参数，工具真正去执行。

没有工具系统，Agent 和聊天机器人没有本质区别。

dsh 的工具系统不是一个简单的函数注册表。它承担了从"模型决定调哪个函数"到"结果安全地回到模型"这整条链路上的所有工程问题：

模型需要 JSON Schema 才能知道工具参数格式——dsh 从你的 TypeScript 类型自动生成
工具执行可能有风险——dsh 提供三层流水线拦截
有些工具操作不可逆——dsh 内置权限审批机制，危险操作暂停等用户确认
不同 Agent 需要不同工具子集——dsh 支持细粒度的 Scope 隔离

这篇文章把这些机制逐一讲清楚。

`ctx.tools` 是什么

ctx.tools 是一个 Cordis Service，类型是 ToolRuntime。它是工具系统的入口，负责：

注册：管理全局工具表，支持作用域（Scope）级别的工具覆盖
Schema 投影：把内部定义投影成模型能看懂的 ToolSchema[]
执行：驱动三阶段流水线，返回类型化的 ToolExecutionResult

使用它需要在插件中声明依赖：

<span>// 插件声明</span>
<span>export</span> <span>const</span> inject = [<span>'tools'</span>]

<span>export</span> <span>function</span> <span>apply</span>(<span>ctx: Context</span>): <span>void</span> {
  <span>// 注册工具</span>
  ctx.<span>tools</span>.<span>register</span>(myTool)
  
  <span>// 查询当前 Scope 可见的 Schema（传给模型）</span>
  <span>const</span> schemas = ctx.<span>tools</span>.<span>schemas</span>()
}

插件卸载时，所有通过 ctx.tools.register 注册的工具会自动撤销——这是 Cordis Effect 机制的应用（见第二篇）。

`defineTool`：类型安全的工具定义

直接实现 ToolDefinition 接口是可以的，但 dsh 提供了更便捷的 defineTool 辅助函数，它在编译期做类型推断，并在运行时自动校验参数和返回值。

一个最小示例：

<span>import</span> { defineTool } <span>from</span> <span>'@deepseek-ai/dsh-tools'</span>

<span>const</span> greetTool = <span>defineTool</span>({
  <span>// 工具名（模型用这个名字调用）</span>
  <span>name</span>: <span>'greet_user'</span>,
  <span>// 工具描述（模型根据这个决定什么时候用）</span>
  <span>description</span>: <span>'Send a greeting to a user by name.'</span>,
  
  <span>// 参数 Schema：使用 dsh 的 DSL，不是原始 JSON Schema</span>
  <span>parameters</span>: {
    <span>name</span>: {
      <span>type</span>: <span>'string'</span>,
      <span>required</span>: <span>true</span>,             <span>// required: true 表示必填</span>
      <span>description</span>: <span>'The user name to greet.'</span>,
    },
    <span>formal</span>: {
      <span>type</span>: <span>'boolean'</span>,
      <span>description</span>: <span>'Use formal greeting if true.'</span>,
      <span>// 没有 required: true = 可选参数</span>
    },
  },
  
  <span>// 输出定义：声明返回值的 Schema 和如何渲染给模型</span>
  <span>output</span>: {
    <span>schema</span>: { <span>type</span>: <span>'string'</span> },
    <span>render</span>: <span>(<span>_args, value</span>) =></span> [{ <span>type</span>: <span>'text'</span>, <span>text</span>: value <span>as</span> <span>string</span> }],
  },
  
  <span>// 执行函数：args 类型由 parameters 自动推断</span>
  <span>async</span> <span>execute</span>(<span>args, exec</span>) {
    <span>// TypeScript 知道 args.name: string，args.formal?: boolean</span>
    <span>const</span> prefix = args.<span>formal</span> ? <span>'Good day'</span> : <span>'Hello'</span>
    <span>return</span> <span>`<span>${prefix}</span>, <span>${args.name}</span>!`</span>
  },
})

defineTool 干了三件事：

从 parameters 声明推断 args 的 TypeScript 类型——不需要手写类型标注
把参数 Schema DSL 编译成 JSON Schema——自动发送给模型
运行时校验：参数不符合 Schema 抛 ToolArgsError，返回值不符合 output.schema 抛 ToolOutputError

Schema DSL 类型系统

dsh 的 Schema DSL（ValueSchemaSpec）支持以下节点类型：

<span>// 字符串</span>
{ <span>type</span>: <span>'string'</span>, <span>enum</span>?: <span>string</span>[], <span>const</span>?: <span>string</span> }

<span>// 数字</span>
{ <span>type</span>: <span>'number'</span> }
{ <span>type</span>: <span>'integer'</span> }

<span>// 布尔</span>
{ <span>type</span>: <span>'boolean'</span> }

<span>// 数组</span>
{ <span>type</span>: <span>'array'</span>, items?: <span>ValueSchemaSpec</span> }

<span>// 对象（必须声明 additionalProperties）</span>
{ <span>type</span>: <span>'object'</span>, properties?: <span>ParameterSchemaSpec</span>, <span>additionalProperties</span>: <span>boolean</span> }

<span>// 无约束 JSON（任意 JSON 值）</span>
{ <span>type</span>: <span>'json'</span> }

<span>// 精确匹配一个分支</span>
{ <span>oneOf</span>: [spec1, spec2, ...] }

**为什么不直接用原始 JSON Schema？**原始 JSON Schema 太宽泛，很多关键字 dsh 并不支持。用 dsh DSL，编译时能确保你声明的约束是 dsh 真正会执行的约束——没有"写了但不生效"的死角。

类型推断深度到 16 层容器，超过后回退到 JsonValue——这个设计避免 TypeScript 类型实例化栈溢出。

三阶段执行流水线

这是工具系统最重要的部分。当 Agent Loop 收到模型的工具调用请求，它会走如下流水线：

模型请求 "调用 shell_run(cmd='ls -la')"
           │
           ▼
    ┌─────────────────┐
    │  tools<span>/</span>pre<span>-</span><span>execute</span> │  waterfall：allow <span>/</span> deny <span>/</span> ask
    └─────────────────┘
           │ allow（通过）
           ▼
    ┌─────────────────┐
    │    Guards       │  单调拒绝检查（只能拒，不能放行）
    └─────────────────┘
           │ 通过
           ▼
    ┌─────────────────┐
    │  tools<span>/</span><span>execute</span>  │  waterfall：around<span>-</span>dispatch（超时<span>/</span>重试<span>/</span>指标）
    └─────────────────┘
           │
           ▼ 工具 <span>execute</span>() 函数运行
           │
    ┌─────────────────┐
    │ tools<span>/</span>post<span>-</span><span>execute</span> │  waterfall：accept <span>/</span> replace <span>/</span> block
    └─────────────────┘
           │
           ▼
    ┌─────────────────┐
    │  finalizeContent │  工具自有的最终内容调整（可选）
    └─────────────────┘
           │
           ▼
    ┌─────────────────┐
    │  tools<span>/</span><span>result</span>   │  emit：只读观察，用于日志<span>/</span>监控
    └─────────────────┘
           │
           ▼
     返回 ToolExecutionResult 给 Agent Loop

三个核心阶段：

阶段一：`tools/pre-execute`（执行前）

这里决定一个工具调用允不允许运行。监听器返回三种决策：

<span>type</span> <span>PreToolDecision</span> =
  | { <span>kind</span>: <span>'allow'</span> }               <span>// 允许执行</span>
  | { <span>kind</span>: <span>'deny'</span>; <span>reason</span>: <span>string</span> } <span>// 拒绝，并告知原因（传给模型）</span>
  | { <span>kind</span>: <span>'ask'</span>; reason?: <span>string</span> } <span>// 暂停，请求用户审批</span>

ask 是 dsh 权限系统的核心。当监听器返回 ask，dsh 会调用 ctx.get('approval') 服务，等用户确认后才继续。如果没有审批服务，ask 自动变为 deny。

一个常见用例：给 shell 命令工具加权限拦截：

<span>// 在插件里监听 tools/pre-execute</span>
ctx.<span>on</span>(<span>'tools/pre-execute'</span>, <span>async</span> (exec, next) => {
  <span>// 只拦截 shell_run 工具</span>
  <span>if</span> (exec.<span>name</span> !== <span>'shell_run'</span>) <span>return</span> <span>next</span>()
  
  <span>const</span> cmd = (exec.<span>arguments</span> <span>as</span> { <span>command</span>: <span>string</span> }).<span>command</span>
  
  <span>// 包含危险命令的操作需要用户确认</span>
  <span>if</span> (cmd.<span>includes</span>(<span>'rm'</span>) || cmd.<span>includes</span>(<span>'sudo'</span>)) {
    <span>return</span> { <span>kind</span>: <span>'ask'</span>, <span>reason</span>: <span>`Will run: <span>${cmd}</span>`</span> }
  }
  
  <span>return</span> <span>next</span>() <span>// 其他命令直接允许</span>
})

注意：tools/pre-execute 是 waterfall，和第二篇讲的一样，监听器必须要么调用 next() 要么返回一个决策。不调用 next() 会截断后续所有监听器。

阶段二：`tools/execute`（环绕分派）

这是工具函数体真正运行的地方。监听器套在工具执行外面，适合做：

<span>// 添加超时控制</span>
ctx.<span>on</span>(<span>'tools/execute'</span>, <span>async</span> (exec, next) => {
  <span>const</span> timer = <span>new</span> <span>Promise</span><<span>ToolExecutionResult</span>>(<span>(<span>_, reject</span>) =></span>
    <span>setTimeout</span>(<span>() =></span> <span>reject</span>(<span>new</span> <span>Error</span>(<span>'Tool timeout'</span>)), <span>30_000</span>)
  )
  <span>return</span> <span>Promise</span>.<span>race</span>([<span>next</span>(), timer])
})

<span>// 添加执行时间指标</span>
ctx.<span>on</span>(<span>'tools/execute'</span>, <span>async</span> (exec, next) => {
  <span>const</span> start = performance.<span>now</span>()
  <span>const</span> result = <span>await</span> <span>next</span>()
  <span>console</span>.<span>log</span>(<span>`<span>${exec.name}</span> took <span>${performance.now() - start}</span>ms`</span>)
  <span>return</span> result
})

监听器只能修改 exec.signal（取消信号），不能修改工具参数——参数此时已经记录在日志里，改了会导致日志和实际执行不一致。

阶段三：`tools/post-execute`（执行后）

工具执行完成后，这里决定把什么结果给模型：

<span>type</span> <span>PostToolDecision</span> =
  | { <span>kind</span>: <span>'accept'</span> }                              <span>// 直接接受原始结果</span>
  | { <span>kind</span>: <span>'accept'</span>; <span>content</span>: <span>ContentBlock</span>[] }     <span>// 替换展示内容（保留规范值）</span>
  | { <span>kind</span>: <span>'accept'</span>; <span>value</span>: <span>JsonValue</span> }            <span>// 替换规范值（重新渲染内容）</span>
  | { <span>kind</span>: <span>'block'</span>; <span>feedback</span>: <span>ContentBlock</span>[] }     <span>// 改为错误结果，附上纠正反馈</span>

block 适合做输出校验——如果工具返回了不符合预期的内容，拦截并让模型重试：

ctx.<span>on</span>(<span>'tools/post-execute'</span>, <span>async</span> (exec, result, next) => {
  <span>if</span> (exec.<span>name</span> === <span>'read_file'</span> && !result.<span>isError</span>) {
    <span>// 文件内容太长，截断后告知模型</span>
    <span>const</span> content = result.<span>value</span> <span>as</span> <span>string</span>
    <span>if</span> (content.<span>length</span> > <span>50_000</span>) {
      <span>return</span> {
        <span>kind</span>: <span>'accept'</span>,
        <span>content</span>: [{ <span>type</span>: <span>'text'</span>, <span>text</span>: <span>`[Truncated] <span>${content.slice(<span>0</span>, <span>50_000</span>)}</span>...`</span> }],
      }
    }
  }
  <span>return</span> <span>next</span>()
})

Guards：单调拒绝

ctx.tools.guard() 注册的 Guard 在 tools/pre-execute 之后、工具函数体之前运行。它只能拒绝，不能放行——即使之前的 waterfall 允许了某个调用，Guard 仍然可以最终否决它：

<span>// Guard 只返回 string（拒绝原因）或 undefined（不干预）</span>
ctx.<span>tools</span>.<span>guard</span>(<span>(<span>exec</span>) =></span> {
  <span>// 禁止在任何工具调用中包含注入模式</span>
  <span>const</span> args = <span>JSON</span>.<span>stringify</span>(exec.<span>arguments</span>)
  <span>if</span> (args.<span>includes</span>(<span>'ignore previous instructions'</span>)) {
    <span>return</span> <span>'Potential prompt injection detected'</span>
  }
  <span>// 返回 undefined = 不干预</span>
})

Guard 的单调性保证：一旦某个 Guard 拒绝了调用，后续任何 Guard 都无法撤销这个拒绝。这防止了恶意插件通过注册 Guard 给自己开后门。

作用域隔离（ToolRestriction）

不同的 Agent 任务可能需要不同的工具子集。比如代码审查 Agent 只需要只读文件工具，不应该能执行 shell 命令。

ctx.tools.restrict() 允许在 Agent 作用域级别过滤工具：

<span>// 只允许文件读取工具</span>
<span>const</span> disposer = ctx.<span>tools</span>.<span>restrict</span>({
  <span>allow</span>: [<span>'read_file'</span>, <span>'list_files'</span>, <span>'search_files'</span>],
})

<span>// 或者：禁用危险工具</span>
<span>const</span> disposer = ctx.<span>tools</span>.<span>restrict</span>({
  <span>deny</span>: [<span>'shell_run'</span>, <span>'write_file'</span>, <span>'delete_file'</span>],
})

<span>// 清理时撤销限制</span>
<span>disposer</span>()

作用域中，allow 和 deny 取交集：如果有多个 restrict 调用，只有所有规则都允许的工具才能见到模型。

作用域自身注册的工具不受限制影响——子 Agent 注册的私有工具始终对自己可见。

并行工具调用

模型有时会在一条消息中请求多个工具调用。dsh 支持并行执行，但需要工具明确声明自己是并发安全的：

<span>const</span> safeReadTool = <span>defineTool</span>({
  <span>name</span>: <span>'read_file'</span>,
  <span>// ...</span>
  <span>// 只读操作，可以和其他调用并行</span>
  <span>isConcurrencySafe</span>: <span>(<span>args</span>) =></span> <span>true</span>,
  <span>execute</span>: <span>async</span> (args, exec) => {
    <span>return</span> fs.<span>readFile</span>(args.<span>path</span>, <span>'utf-8'</span>)
  },
})

没有声明 isConcurrencySafe 或返回非 true 的工具，默认以独占（exclusive）模式运行——一次只跑一个，前一个完成后才开始下一个。

这个设计是防御性的：宁可串行慢点，也不让状态竞争。声明并发安全是工具作者的承诺，不是框架帮你猜的。

实战：从零写一个天气查询工具

把上面所有概念串起来，写一个完整的自定义工具：

<span>// packages/my-tools/src/weather.ts</span>
<span>import</span> <span>type</span> { <span>Context</span> } <span>from</span> <span>'@deepseek-ai/cordis'</span>
<span>import</span> { defineTool } <span>from</span> <span>'@deepseek-ai/dsh-tools'</span>

<span>export</span> <span>const</span> name = <span>'tool-weather'</span>
<span>export</span> <span>const</span> inject = [<span>'tools'</span>]

<span>export</span> <span>function</span> <span>apply</span>(<span>ctx: Context</span>): <span>void</span> {
  ctx.<span>tools</span>.<span>register</span>(<span>defineTool</span>({
    <span>name</span>: <span>'get_weather'</span>,
    <span>description</span>: <span>'Get current weather for a city. Returns temperature and conditions.'</span>,
    
    <span>parameters</span>: {
      <span>city</span>: {
        <span>type</span>: <span>'string'</span>,
        <span>required</span>: <span>true</span>,
        <span>description</span>: <span>'City name, e.g. "Beijing" or "Shanghai".'</span>,
      },
      <span>unit</span>: {
        <span>type</span>: <span>'string'</span>,
        <span>enum</span>: [<span>'celsius'</span>, <span>'fahrenheit'</span>],
        <span>description</span>: <span>'Temperature unit. Defaults to celsius.'</span>,
      },
    },
    
    <span>output</span>: {
      <span>schema</span>: {
        <span>type</span>: <span>'object'</span>,
        <span>additionalProperties</span>: <span>false</span>,
        <span>properties</span>: {
          <span>city</span>: { <span>type</span>: <span>'string'</span> },
          <span>temperature</span>: { <span>type</span>: <span>'number'</span> },
          <span>unit</span>: { <span>type</span>: <span>'string'</span> },
          <span>condition</span>: { <span>type</span>: <span>'string'</span> },
        },
      },
      <span>// render 把结构化值转成模型看到的文字</span>
      <span>render</span>: <span>(<span>_args, value</span>) =></span> {
        <span>const</span> v = value <span>as</span> { <span>city</span>: <span>string</span>; <span>temperature</span>: <span>number</span>; <span>unit</span>: <span>string</span>; <span>condition</span>: <span>string</span> }
        <span>return</span> [{
          <span>type</span>: <span>'text'</span>,
          <span>text</span>: <span>`Weather in <span>${v.city}</span>: <span>${v.temperature}</span>°<span>${v.unit === <span>'celsius'</span> ? <span>'C'</span> : <span>'F'</span>}</span>, <span>${v.condition}</span>`</span>,
        }]
      },
    },
    
    <span>// 只读操作，允许并行</span>
    <span>isConcurrencySafe</span>: <span>() =></span> <span>true</span>,
    
    <span>async</span> <span>execute</span>(<span>args, exec</span>) {
      <span>const</span> unit = args.<span>unit</span> ?? <span>'celsius'</span>
      
      <span>// 真实场景这里调用天气 API</span>
      <span>// exec.signal 用于响应取消</span>
      <span>const</span> response = <span>await</span> <span>fetch</span>(
        <span>`https://weather-api.example.com/current?city=<span>${<span>encodeURIComponent</span>(args.city)}</span>&unit=<span>${unit}</span>`</span>,
        { <span>signal</span>: exec.<span>signal</span> },
      )
      
      <span>if</span> (!response.<span>ok</span>) {
        <span>throw</span> <span>new</span> <span>Error</span>(<span>`Weather API returned <span>${response.status}</span>`</span>)
      }
      
      <span>const</span> data = <span>await</span> response.<span>json</span>()
      <span>return</span> {
        <span>city</span>: args.<span>city</span>,
        <span>temperature</span>: data.<span>temperature</span>,
        unit,
        <span>condition</span>: data.<span>condition</span>,
      }
    },
    
    <span>// 自定义 UI 展示：调用中显示什么</span>
    <span>presentCall</span>: <span>(<span>args</span>) =></span> ({
      <span>card</span>: <span>'generic'</span>,
      <span>title</span>: <span>`Checking weather in <span>${(args <span>as</span> { city: <span>string</span> }).city}</span>`</span>,
      <span>kind</span>: <span>'fetch'</span>,
    }),
    
    <span>// 自定义 UI 展示：完成后显示什么</span>
    <span>presentResult</span>: <span>(<span>args, result</span>) =></span> ({
      <span>card</span>: <span>'generic'</span>,
      <span>title</span>: result.<span>isError</span>
        ? <span>`Weather check failed`</span>
        : <span>`Weather in <span>${(args <span>as</span> { city: <span>string</span> }).city}</span>`</span>,
    }),
  }))
}

注册到 dsh 配置：

<span># dsh.config.yml（或你的 profile）</span>
<span>plugins:</span>
  <span>-</span> <span>id:</span> <span>tool-weather</span>
    <span># 指向你的包路径</span>

就这些。工具注册后：

ctx.tools.schemas() 会自动包含 get_weather 的 JSON Schema，模型能看到它
模型调用时自动经过完整三阶段流水线
返回值自动校验、渲染、记入 Session 日志

架构设计总结

回顾一下 dsh 工具系统的整体设计哲学：

设计决策	背后的原因
参数在 pre-execute 后不可修改	日志、审计、UI 展示必须和执行保持一致
Guard 只能拒绝，不能放行	防止后挂插件撤销安全策略
output.render 是纯函数	Session 回放时需要重现展示效果
isConcurrencySafe 默认独占	防御性设计，状态安全优先于性能
Tools 是 Cordis Service	注册/注销的生命周期跟随插件，HMR 不泄露工具
工具 Schema 不泄露 execute/present 等回调	模型只看 name/description/parameters

dsh 的工具系统设计有一个贯穿始终的原则：在工具调用的每个阶段，都有且只有一个清晰的扩展点，并且扩展点的职责边界明确——不能越界。

这就是为什么你能在生产环境放心地挂上权限拦截器，而不用担心它和某个内部逻辑冲突。

系列下一篇

工具系统解决了"Agent 能做什么"的问题。下一篇 Agent Loop 会讲"一次对话是怎么跑起来的"——模型输出、工具调用、循环终止的完整流程，以及你可以在哪些地方插入控制逻辑。

在 PrimeSkills 可以找到已在真实企业场景验证过的 AI Agent 技能和工作流，不是演示级的，是用在实际项目里的。

更多内容见我的个人主页