青雲的博客
深入浅出 DeepSeek Harness 第七部：产品面——同一运行时的不同面孔 第 41 章

TypeScript 和 Python SDK 共享一套协议

SDK 不是简单 API wrapper——TypeScript 和 Python SDK 共享 JSON-RPC 协议契约，封装连接管理、session 生命周期、事件订阅、streaming 处理。SDK consumer 看不到 Cordis/AgentLoop/Session 实现细节。

源码版本
47f943859bef60e4160492346772ded9b24f765a
验证日期
Commit
47f943859bef60e4160492346772ded9b24f765a

你可能觉得 SDK 就是”把 JSON-RPC 调用包装成函数”的薄皮。如果只是那样，TS 和 Python 各写各的就行，不需要专门讲一章。

DSH 的 SDK 设计不一样。TS SDK 和 Python SDK 不是两个独立实现各自拼协议——它们共享同一套协议契约，以 packages/sdk/protocol/src/types.ts 为唯一类型源头。TS 侧直接 import 这些类型，Python 侧用 Pydantic model 声明等价的 shape（字段名、类型、结构完全对应）。协议一致性靠类型约束+集成测试保证，不靠手写文档。

更关键的是：SDK 做的事情比”发 HTTP 请求”多得多。它要管理子进程生命周期、处理 transport 断开重连、维护 notification 队列和订阅过滤、追踪子 agent session 树、实现 EOF→SIGTERM→SIGKILL 的优雅退出梯。这些是 SDK consumer 完全看不到的。

SDK 架构三层：protocol → client → high-level API

SDK 代码分三个包，TS 和 Python 各对应：

第一层是 @deepseek-ai/dsh-sdk-protocol（Python 侧是 models.py），定义 wire 上的类型。这是唯一的协议源头。它不做 IO，不做逻辑，只做类型声明和 transport 编解码。

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
sed -n '100,105p' "$repo/packages/sdk/protocol/src/types.ts"

第二层是 low-level client：TS 的 HarnessClient（packages/sdk/client/src/client.ts），Python 的 HarnessClient（python/sdk/src/deepseek_harness/client.py）。这层负责 spawn 子进程、读写 stdio、序列化/反序列化 JSON-RPC、管理 request/response 映射、分发 notification 给订阅者、处理超时和进程退出。

第三层是 high-level API：TS 侧直接用 HarnessClient（它本身就提供了 subscribe/prompt/initialize 的完整接口），Python 侧在 HarnessClient 上又包了一层 DeepSeekHarness + Session 类，提供 harness.run("task") 这种一次性接口，帮你自动收集 events 和 final response。

TS HarnessClient：子进程管理与 notification 队列

TS 的 HarnessClient 构造时接收 options（command、args、cwd、env、requestTimeoutMs、shutdownTimeoutMs 等），不立即 spawn 子进程。start() 懒启动：spawn(command, args, { stdio: ['pipe','pipe','pipe'] })，然后创建 JsonRpcLineTransport 挂在 child.stdout/child.stdin 上，开始读帧。

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
sed -n '203,261p' "$repo/packages/sdk/client/src/client.ts"

start 后，client 监听几个关键事件：

  • child.on('error')：spawn 失败，transport.close() + fail 所有 subscription
  • child.stdin.on('error')：写 EPIPE 非致命处理（race with death）
  • child.stderr.on('data')：收集 stderr 行到 ring buffer（最多 400 行，诊断用）
  • child.on('exit')：记录 exitCode，fail 所有 subscription
  • child.on('close')：transport.close()，所有 pending request 失败

notification 分发用的是订阅模式。subscribe(filter?) 返回一个 `NotificationSubscription，内部有 queue + waiters 结构。notification 到达时，如果有 waiters（有人在 await next()）就直接 resolve；没有就推入 queue（有界消费者）。filter 抛错只 fail 这个 subscription，不影响其他订阅者和 transport 读循环——和 Python SDK 行为一致。

flowchart LR
    subgraph SDK Process
        C["HarnessClient"]
        T["JsonRpcLineTransport"]
        Q["Notification Queue"]
        W["Waiters (Promise)"]
    end

    subgraph Runtime Subprocess
        R["dsh-runtime"]
        S["JSON-RPC Server"]
        A["Agent(s)"]
    end

    C -->|"spawn"| R
    C -->|"stdin.write(JSON-RPC)"| T
    T -->|"stdout frames"| C
    C -->|"dispatch"| Q
    Q -->|"resolve"| W
    R --> S --> A

    C -->|"initialize/prompt/shutdown"| T
    S -->|"session.event/status\nsubagent.* notifications"| T

    style C fill:#581c87,stroke:#a855f7,color:#fff
    style R fill:#1e3a5f,stroke:#3b82f6,color:#fff

Python HarnessClient：线程读环 + 队列模型

Python SDK 因为 GIL 和同步 IO 模型，用了不同的并发结构但同一套协议逻辑。Python HarnessClient.init 创建 _responses: dict[str, queue.Queue]（按 request id 映射 response 队列）、_notifications: queue.Queue（全局 notification 队列）、_notification_subscribers: dict[str, tuple[Queue, filter]]（按订阅 id 映射队列）。

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
sed -n '37,55p' "$repo/python/sdk/src/deepseek_harness/client.py"

Python 版本用两个 daemon 线程：_start_reader_thread() 从 stdout 逐行读 JSON-RPC frame，根据有没有 id 分发给 response queue 或 notification 队列/订阅者；_start_stderr_thread() 读 stderr 到 deque(maxlen=400)。TS 版本用的是 EventEmitter + async iterator，Python 用 threading + queue.Queue——但对 SDK 使用者来说，接口形状完全对称：start()、initialize()、session_prompt()、subscribe_session_notifications()、close()。

Python 还在 low-level HarnessClient 上面包了 DeepSeekHarness 和 Session 两个高级类（api.py）。DeepSeekHarness.run(input, session_id=None) 自动 start+initialize、创建 Session、订阅 session 树 notification、等 session.status 变 idle、收集 events、提取 final_response 和 finish_reason——这是给”一次性 run”场景用的便利 API。TS SDK 没有这个包装（TS 侧直接用 HarnessClient 自己控制），但你可以很容易自己实现同样的逻辑。

subscribeSessionTree：客户端追踪子 agent 树

SDK 的一个巧妙设计是 subscribeSessionTree(sessionId)（TS）/ subscribe_session_notifications(session_id)（Python）。server 端的 subagent.started notification 带 parentSessionId 和 childSessionId。客户端收到后维护一个 sessionParents: Map<string, string>（childId→parentId），订阅 filter 时用 isDescendantOf 递归查 parent 链，判断一个事件是否属于指定 session 或其子 agent 树。

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
sed -n '361,430p' "$repo/packages/sdk/client/src/client.ts"

这个追踪完全在客户端做。server 不做过滤——它把所有 session 的 event 都推过来（因为它不知道客户端关心哪些），客户端靠 parent 链自己决定哪些属于”我的树”。这意味着：如果你只 subscribe root session 的 event，会错过子 agent 的事件；用 subscribeSessionTree 才能拿到完整的 agent 树活动。

close：EOF→SIGTERM→SIGKILL 退出梯

两个 SDK 的 close 都遵循同一个退出梯：先发 protocol `shutdown 请求（bounded by shutdownTimeoutMs），然后关 stdin（EOF），等进程自己退出（disposeEofGraceMs，默认 6 秒），没退就 SIGTERM（disposeGraceMs，默认 3 秒），还没退就 SIGKILL。

repo="${DSH_SOURCE_DIR:?set DSH_SOURCE_DIR to the official fixed checkout}"
sed -n '385,401p' "$repo/packages/sdk/client/src/client.ts"

注意：request timeout 用 AbortController 做 abandon，不是 cancel。TS 代码注释明确说 “there is no wire-level cancel: a timed-out request stays running server-side until the runtime is closed”。超时只是客户端不再等结果，server 端的工作继续跑到自然结束（或 shutdown）。

容易踩的坑

坑一：把 SDK 当成 HTTP REST client。 SDK 默认通过 stdio spawn 子进程通信，不是 HTTP。你不能用 Postman 或 curl 测试 SDK server。transport 层也可以切换（比如 TCP），但 out of the box 是子进程 stdio。如果你想从另一个语言接入 DSH，要么 spawn 子进程实现 JSON-RPC over stdio，要么等 TCP transport 暴露。

坑二：以为 SDK 里有 Cordis 或 Agent 对象。 SDK consumer 看不到任何内部类型。你拿不到 Agent 引用、拿不到 Session 对象、拿不到 Cordis Context。SDK 只暴露 wire 上定义的形状（InitializeParams、SessionEvent、SubagentStartedNotification 等）。如果你的代码需要直接操作 Agent（比如自定义 tool 注册），SDK 不是正确的入口——你应该直接用 dsh 的 Cordis composition API。

坑三：Python SDK 的 Session.run() 是同步阻塞的。 session.run(input) 会阻塞当前线程直到 session.status 变 idle。TS 版本的 subscribeSessionTree 返回 AsyncIterable，你可以 for-await-of 逐帧消费。Python 如果你需要 streaming 效果，用 on_notification 回调。

坑四：initialize 只做一次。 DeepSeekHarness.start() 里调了 initialize()，之后的 run() 调用复用同一个 runtime 进程和同一个 session 配置（cwd/provider/model）。如果你需要不同 cwd，要 new 一个新的 DeepSeekHarness 实例。SDK server 的 initialize 在 server 实例生命周期内只接受一次——第二次调 initialize 会覆盖 cwd/provider/model，但这不是预期用法。

SDK 连接的是 JSON-RPC Server。但当你打开浏览器用 dsh web 时，浏览器不是 SDK client，它连的是另一个完全不同的入口——ApiProxy。