返回博客列表

用 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_URL 和 DEEPSEEK_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_reason(completed / 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.render:schema 定义 canonical JSON 值的类型,execute 只返回这个值;registry 会 snapshot、校验、freeze,然后传给 render 生成模型可见的内容块。不要从 body 返回 content blocks,不要让调用方解析散文来找 id。
  • execute(args, exec):实际逻辑。args 是从 schema 推导的强类型,exec 带不可变身份信息(callId、name、arguments、agent、token、signal)。

举个 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.token。callId、name、arguments、agent、token、signal 在整个 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_status、deploy_service、analyze_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人工智能时代(the-ai-era) 网站: https://www.theaiera.top/ 关注每日最新AI新闻和技术博客,主页有更多的文章的AI 技术参考:https://www.theaiera.top

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

分享给朋友