返回博客列表

用 DeepSeek Harness Python SDK 定制 DevOps Agent:从 minimal 到自定义工具

2026-08-19T23:30:00+08:00
DeepSeek HarnessPython SDKDevOps AgentCordisAgent 架构Tool自定义工具

用 DeepSeek Harness Python SDK 定制 DevOps Agent:从 minimal 到自定义工具

跟着做一遍,5 分钟搞定。搞不定评论区问我。

上一篇讲了 DeepSeek Harness 的"一切皆插件"架构。这次回答一个更实际的问题:怎么用它的 Python SDK 做一个自己的 DevOps Agent?

官方文档给了 minimal 示例(一个能跑的 30 行脚本),但没告诉你怎么往前走一步--加自定义工具、改行为策略、做成生产级。这篇文章从 minimal 起步,一路拆到自定义工具的完整 API 契约。

本文提纲

  1. SDK 全景:它在跑什么
  2. 30 行跑起来:minimal 示例
  3. Cordis YAML:行为的真正配置层
  4. 自定义工具:defineTool 的完整契约
  5. execute() 的六条铁律
  6. 执行策略钩子:别把策略塞进工具
  7. 长任务:background job 模式
  8. 配一个 DevOps Agent 的实战配方

SDK 全景:它在跑什么

先搞清楚 Python SDK 到底是什么。它不是纯 Python 实现--而是一个驱动子进程的 JSON-RPC 客户端。SDK 安装时会拉一个 deepseek-harness-runtime-bin 的平台 wheel,里面打包了完整的 TypeScript runtime(单文件 dsh-jsonrpc-agent 可执行文件),目标机器不需要装 Node.js

python -m pip install deepseek-harness-sdk
from deepseek_harness import DeepSeekHarness

with DeepSeekHarness() as harness:
    result = harness.run("Say hi.")

DeepSeekHarness 会懒启动一个 runtime 子进程,跨多次调用复用,上下文管理器退出时关闭。runtime 继承 DEEPSEEK_BASE_URLDEEPSEEK_API_KEY 等环境变量,所以你可以直接接真实模型,也可以指向本地 proxy。

不传 cordis 参数时,SDK 注入一份默认组合(stdio JSON-RPC server + agent core + DeepSeek 适配器 + JSONL 会话持久化 + local bash)。要做自定义,传你自己的 Cordis YAML 路径即可,但必须保留 @deepseek-ai/dsh-sdk-jsonrpc-server 这个入口。

30 行跑起来:minimal 示例

官方的 minimal 示例长这样:

from pathlib import Path
from deepseek_harness import DeepSeekHarness

config = Path("examples/jsonrpc-agent/minimal.cordis.yml").resolve()
workspace = Path("/absolute/path/to/workspace").resolve()
sessions = Path("/absolute/path/to/sessions").resolve()

with DeepSeekHarness(
    provider="deepseek-official",
    model="deepseek-v4-flash",
    max_tokens=49_152,
    cwd=str(workspace),
    session_root=str(sessions),
    cordis=str(config),
) as harness:
    result = harness.run(
        "Inspect the repository and fix the failing tests.",
        session_id="example-001",
    )
    print(result.final_response)

构造参数:

参数 作用
provider API provider 路由,如 "deepseek-official"
model 模型 ID,由 provider adapter 解析
max_tokens 每次请求的输出 token 上限
cwd Agent 工作区路径
session_root 会话日志存储路径
cordis Cordis 组合 YAML 文件路径

harness.run() 返回 RunResult,核心字段是 .final_response(最后一次 assistant 回复)。还有 .finish_reasoncompleted / max-tokens / error / None)、.events(root session 事件)、.notifications(root + 所有已知子 agent 通知,按 wire 顺序)。

一个关键行为:复用 session_id 会保留会话专属的 Bash 进程,包括工作目录、exported 变量、shell 函数。独立任务用新 session_id,延续对话才复用。

Cordis YAML:行为的真正配置层

minimal 示例之所以"minimal",不在于 Python 代码,而在于它指向的 minimal.cordis.yml。这个 YAML 才是行为的真正配置层。来看它的完整结构:

- id: sdk-jsonrpc-server
  name: '@deepseek-ai/dsh-sdk-jsonrpc-server'
  config:
    maxTokensAsSuccess: false

- id: llm-deepseek
  name: '@deepseek-ai/dsh-llm-deepseek'
  config:
    apiKeyEnv: DEEPSEEK_API_KEY
    streamIdleTimeoutMs: 172800000
    models:
      - id: !!js process.env.DSH_MODEL ?? 'deepseek-v4-flash'
        contextWindow: !!js Number(process.env.DSH_CONTEXT_WINDOW ?? 1000000)

- id: sandbox
  name: '@deepseek-ai/dsh-sandbox-local'

- id: sandbox-policy
  name: '@deepseek-ai/dsh-sandbox-policy'
  config:
    mode: danger-full-access
    workspaceRoot: !!js process.env.DSH_CWD ?? process.cwd()

- id: pty
  name: '@deepseek-ai/dsh-terminal'

- id: terminal-bash
  name: '@deepseek-ai/dsh-terminal-bash'
  config:
    timeoutMs: 300000

- id: fs-local
  name: '@deepseek-ai/dsh-fs-local'
  config:
    cwd: !!js process.env.DSH_CWD ?? process.cwd()

- id: agent-spine
  name: '@deepseek-ai/dsh-agent-spine-demo'
  config:
    includeHarnessIdentity: false
    includeRuntimeContext: false
    persona: !!js process.env.DSH_SYSTEM_PROMPT ?? 'You are a helpful software engineer assistant.'
    workspaceContext: false
    skills:
      enabled: false
    toolBash: false
    toolJobs: false

- id: persistent-bash
  name: '@deepseek-ai/dsh-tool-bash-persistent'
  config:
    timeoutMs: 300000

- id: str-replace-editor
  name: '@deepseek-ai/dsh-tool-str-replace-editor'
  config:
    maxOutputChars: 16000

- id: sessions
  name: '@deepseek-ai/dsh-session-persistence-jsonl'
  config:
    root: !!js process.env.DSH_SESSION_ROOT ?? './.sessions'
    compression: none

每个条目是一个插件,id 是本地标识,name 是 npm 包名,config 是插件配置。!!js 是 Cordis 的表达式插值--在插件激活后、对当前插件上下文求值,所以可以直接读 process.env

这份配置告诉了我们一个事实:minimal 之所以 minimal,是因为它关掉了 harness identity、workspace prompt、skills、one-shot bash、task tools、compaction。模型只看到 system prompt + 持久 bash + str_replace_editor 两个工具。

想加功能?加对应插件条目。想做 DevOps agent?从这里改起。

自定义工具:defineTool 的完整契约

这是文章的核心。自定义一个模型可见的工具,最小骨架是这样:

import { readFile } from 'node:fs/promises'
import type { Context } from '@deepseek-ai/cordis'
import { defineTool } from '@deepseek-ai/dsh-tools'

export const name = 'my-tool'
export const inject = ['tools']

export function apply(ctx: Context) {
  ctx.tools.register(defineTool({
    name: 'read_file',
    description: 'Read a file from disk.',
    parameters: {
      path: { type: 'string', required: true, description: 'Absolute path' },
      limit: { type: 'number' },
    },
    output: {
      schema: { type: 'string' },
      render: (_args, value) => [{ type: 'text', text: value }],
    },
    async execute(args, exec) {
      return readFile(args.path, { encoding: 'utf8', signal: exec.signal })
    },
  }))
}

注册是 effect-based 的--销毁插件 fiber 就自动注销工具,schema 会自动流入 system prompt 组装。

defineTool 的四个核心字段:

  • name + description:模型看到的工具名和描述。description 写得好不好直接决定模型会不会用、用得对不对。
  • parameters:参数 schema,defineTool 会在 execute 前自动做类型校验(类型、required key、literal 约束、嵌套值)。隐式参数根是开放的,显式 object 节点要声明 additionalProperties
  • output.schema + output.renderschema 定义 canonical JSON 值的类型,execute 只返回这个值;registry 会 snapshot、校验、freeze,然后传给 render 生成模型可见的内容块。不要从 body 返回 content blocks,不要让调用方解析散文来找 id。
  • execute(args, exec):实际逻辑。args 是从 schema 推导的强类型,exec 带不可变身份信息(callIdnameargumentsagenttokensignal)。

举个 DevOps 场景的例子--一个"检查 Kubernetes pod 状态"的工具:

export const name = 'k8s-pod-status'
export const inject = ['tools']

export function apply(ctx: Context) {
  ctx.tools.register(defineTool({
    name: 'k8s_pod_status',
    description: 'Check the status of pods in a Kubernetes namespace. Returns pod names, status, restarts, and age.',
    parameters: {
      namespace: { type: 'string', required: true, description: 'Kubernetes namespace' },
      labelSelector: { type: 'string', description: 'Optional label selector, e.g. app=nginx' },
    },
    output: {
      schema: {
        type: 'object',
        properties: {
          pods: {
            type: 'array',
            items: {
              type: 'object',
              properties: {
                name: { type: 'string' },
                status: { type: 'string' },
                restarts: { type: 'number' },
                age: { type: 'string' },
              },
            },
          },
        },
      },
      render: (_args, value) => [{ type: 'text', text: JSON.stringify(value, null, 2) }],
    },
    async execute(args, exec) {
      // args 已经过类型校验: { namespace: string; labelSelector?: string }
      const { KubeExec } = await import('./kube-client')
      const kube = new KubeExec()
      const selector = args.labelSelector ? `-l ${args.labelSelector}` : ''
      const { stdout } = await kube.exec(
        `kubectl get pods -n ${args.namespace} ${selector} -o json`,
        { signal: exec.signal }
      )
      return JSON.parse(stdout)
    },
  }))
}

execute() 的六条铁律

execute 函数有几条不能违反的契约:

1. 参数已校验。 defineTool 会在 execute 前验证模型生成的 arguments。schema 表达不了的约束(非空字符串、正数、跨字段规则)你还得自己检查。

2. 定义注册后不可变。 注册 borrow 的是你的 readonly 定义,不要在注册后改 schema 或替换 callback。要热替换工具,dispose owning effect 再注册新的。

3. 执行身份受保护。 registry 会把 arguments 做成 detached lossless JSON、freeze、赋一个 opaque exec.tokencallIdnameargumentsagenttokensignal 在整个 dispatch 期间不可变。只有 around-dispatch wrapper 能拿到 mutable view,且只能替换/恢复 exec.signal(用于加超时),不能移除它。

4. 返回一个 canonical JSON 值。 execute 只返回 inferred value,registry 做 snapshot、校验、freeze,再传给 render。throw 或返回非法值都算 isError

5. 响应 exec.signal 取消信号触发时,必须 cancel 正在进行的工作。这是 Agent 不卡死的底线。

6. 用 exec.agent 做异步通知。 agent.inject({ content, source: { kind: 'plugin', plugin: '<name>' } }) 会在下一次 model request 注入持久上下文--注意它不是唤醒(idle agent 不会因此动起来)。要防 disposed agent,try/catch 包一层。

执行策略钩子:别把策略塞进工具

文档里有句话特别值得记住:"Prefer not to build deployment policy into the tool."

不要在 execute 里写"是否允许执行"的逻辑。Cordis 提供了专门的执行策略钩子:

钩子 作用
tools/pre-execute 可扩展的 allow/deny/ask 策略(permission gate)
ctx.tools.guard() 最终的单调 deny,后续 listener 无法撤销
tools/execute 包裹 dispatch,加超时、重试、metrics
tools/post-execute 替换展示内容或返回值,阻止结果,附加 model-facing 上下文
tools/result 观察不可变的 normalized outcome

一个 permission gate 插件大概长这样(伪代码):

export function apply(ctx: Context) {
  ctx.on('tools/pre-execute', (event) => {
    if (event.name === 'k8s_pod_status' && !isProductionSafe(event.arguments)) {
      return { decision: 'deny', reason: 'Production namespace requires approval' }
    }
    return { decision: 'allow' }
  })
}

这种分离带来的好处:工具本身只管"怎么做",策略层管"允不允许做"。同一个工具,开发环境全开、生产环境加审批,工具代码一行不改。

长任务:background job 模式

DevOps 场景经常遇到长任务--部署、迁移、日志分析。foreground 执行会阻塞 Agent。解法是 ctx.jobs.start()

async execute(args, exec) {
  const job = await ctx.jobs.start({
    kind: 'deploy',
    label: `Deploy ${args.service}`,
    owner: exec.agent,
    run: async (controller) => {
      // 部署逻辑
      const result = await deploy(args.service, controller.signal)
      controller.done(result)
    },
  })
  // 返回 job handle,模型可以继续做别的事
  return { kind: 'background', jobId: job.id }
}

几个要点:

  • owner: exec.agent 让 job 跟随 agent 生命周期
  • controller.signal 是 job 专属的取消信号,不是 exec.signal--外层 call 取消只停等待,不停已发布的 job
  • job_kill、owner disposal、service teardown 负责 job 生命周期
  • 成功启动返回 { kind: 'background', jobId },模型可以从这个结构化值拿到 job id 继续操作

配一个 DevOps Agent 的实战配方

把上面所有零件拼起来。一个 DevOps Agent 的 Cordis YAML 大概是:

# DevOps Agent composition
- id: sdk-jsonrpc-server
  name: '@deepseek-ai/dsh-sdk-jsonrpc-server'

- id: llm-deepseek
  name: '@deepseek-ai/dsh-llm-deepseek'
  config:
    apiKeyEnv: DEEPSEEK_API_KEY
    models:
      - id: !!js process.env.DSH_MODEL ?? 'deepseek-v4-flash'
        contextWindow: !!js Number(process.env.DSH_CONTEXT_WINDOW ?? 1000000)

# 沙箱策略:dev 环境用 danger-full-access,prod 用 sandbox
- id: sandbox-policy
  name: '@deepseek-ai/dsh-sandbox-policy'
  config:
    mode: !!js process.env.DSH_SANDBOX_MODE ?? 'danger-full-access'
    workspaceRoot: !!js process.env.DSH_CWD ?? process.cwd()

# 持久 bash + 文件编辑(继承自 minimal)
- id: persistent-bash
  name: '@deepseek-ai/dsh-tool-bash-persistent'
  config:
    timeoutMs: 600000  # DevOps 命令可能更久

- id: str-replace-editor
  name: '@deepseek-ai/dsh-tool-str-replace-editor'
  config:
    maxOutputChars: 32000

# Agent spine:启用 skills 和 workspace context
- id: agent-spine
  name: '@deepseek-ai/dsh-agent-spine-demo'
  config:
    persona: |
      You are a DevOps assistant. You can inspect Kubernetes clusters,
      manage deployments, analyze logs, and troubleshoot incidents.
      Always verify the current context (kubectl config current-context)
      before operating on production. Prefer non-destructive operations.
    workspaceContext: true
    skills:
      enabled: true
    toolJobs: true  # 启用 background jobs

# 开启 context compaction(长会话必备)
- id: compaction
  name: '@deepseek-ai/dsh-compaction'
  config:
    enabled: true

# JSONL session 持久化
- id: sessions
  name: '@deepseek-ai/dsh-session-persistence-jsonl'
  config:
    root: !!js process.env.DSH_SESSION_ROOT ?? './.sessions'
    compression: gzip  # 生产环境压缩

然后把你写的 DevOps 专用工具(k8s_pod_statusdeploy_serviceanalyze_logs 等)作为 Cordis 插件注册进去。Python 调用层不变:

from deepseek_harness import DeepSeekHarness

with DeepSeekHarness(
    provider="deepseek-official",
    model="deepseek-v4-flash",
    max_tokens=49_152,
    cwd="/opt/devops-workspace",
    session_root="/var/log/dsh-sessions",
    cordis="compositions/devops-agent.cordis.yml",
) as harness:
    result = harness.run(
        "Check why the payment-service pods keep crashing in production, "
        "analyze the logs, and propose a fix.",
        session_id="incident-2026-0819-001",
    )
    print(result.final_response)
    # result.notifications 包含所有子 agent 和 session 事件

一个完整的 DevOps Agent 骨架就搭起来了:bash 工具执行 kubectl/helm 命令,自定义工具做结构化的 pod 状态查询和日志分析,background job 跑长部署任务,compaction 压缩长会话,permission gate 拦截危险操作。

三个安全提醒

文档里反复强调的安全点,DevOps 场景尤其要听:

danger-full-access 模式下,bash 和 editor 能改 runtime 进程可见的任何路径。 minimal 示例用的就是这个。只能在 disposable checkout 或 container 里跑。DevOps Agent 要操作生产集群,必须换成 sandbox 模式或加 permission gate。

persistent PTY 需要 POSIX 终端环境。 Windows 不支持。CI/CD 环境要确保有 PTY 可用。

virtual_mode=True 在开了 shell 之后不提供任何安全性。 路径沙箱挡不住 shell 里的 cat /etc/shadow。生产环境的正解是 sandbox backend + ctx.tools.guard() 声明式拒绝规则 + HITL 审批。

参考文档与链接


作者: itech001 来源: 公众号:AI人工智能时代 网站: https://www.theaiera.cn/ 每日分享最前沿的AI新闻资讯和技术研究。

本文首发于 AI人工智能时代,转载请注明出处。

分享给朋友