Python SDK：跨语言写 DSH 插件

读完这篇你会

• 在 Python 进程里调 DSH 的 Cordis 服务（不限于 TS）
• 解决 Python � TypeScript 互调的类型声明与异步陷阱
• 写一个 Python-only 工具插件

适用版本

Compatible with: dsh 0.1.1-rc.2（基于官方 packages/host-python/ + docs/user/develop/sdk/ 整理）

什么时候用 Python SDK 而不是写 TS 插件

DSH 的 Cordis 内核是 TS，但 host 暴露了 Python 绑定（@deepseek-ai/dsh-host-python）。典型场景：

• 团队的主力语言是 Python，工具链全 Python（数据科学 / ML / 内部平台）
• 想接公司已有 Python 资产（pandas pipeline / 内部门户的 Python SDK）
• 不想维护 TS 代码但又想给 DSH 加能力

不适用场景：

• 高频路径上的工具（Python 进程间 IPC 有 5-15ms 开销，不如直接 TS）
• 强类型契约复杂的工具（TS 类型系统 + IDE 自动补全远比 Python 强）

Python 包结构

my-python-plugin/
├── pyproject.toml
├── dsh.bundle
└── src/
    └── my_plugin/
        ├── __init__.py
        └── tools.py

pyproject.toml 关键字段：

[project]
name = "my-dsh-plugin"
version = "0.1.0"
dependencies = [
    "deepseek-harness-host-python>=0.1.1",
    "pandas>=2.0",
]

dsh.bundle 同 TS 插件一致：

name: my-dsh-plugin
version: 0.1.0
entry: src.my_plugin.tools:apply
language: python

写一个最小 Python 工具

from deepseek_harness import Context, tool

def apply(ctx: Context):
    @tool(name="load_csv", description="Load a CSV file into a pandas DataFrame")
    def load_csv(path: str) -> dict:
        import pandas as pd
        df = pd.read_csv(path)
        return {
            "rows": len(df),
            "columns": list(df.columns),
            "head": df.head(5).to_dict(orient="records"),
        }

@tool 装饰器把函数注册成模型可调用的 tool，参数类型注解会被转成 JSON Schema 给模型。

类型声明 vs TS 互调

Python 类型 → JSON Schema（模型可见）

• str → {"type": "string"}
• int / float → {"type": "integer" | "number"}
• bool → {"type": "boolean"}
• list[X] → {"type": "array", "items": <X schema>}
• dict[str, X] → {"type": "object", "additionalProperties": <X schema>}
• Optional[X] → [X schema, {"type": "null"}]

复杂类型（pydantic BaseModel / dataclass）会自动展开；不要用 Any。

TS 调用 Python 插件

反过来也支持——TypeScript 写的 host 可以 @tool 调用 Python 插件的工具，运行时通过 IPC 把 call 路由到 Python 进程。

异步陷阱

DSH 的工具调用是 async 的。Python SDK 用 asyncio + aiohttp：

import aiohttp

@tool(name="fetch_url", description="Fetch a URL and return text content")
async def fetch_url(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return {"status": resp.status, "text": await resp.text()}

不要在 @tool 函数里写同步阻塞调用（requests.get / time.sleep / pd.read_csv 超大文件）——会卡住整个 agent 循环。CPU 密集型用 asyncio.to_thread() 包一层。

故障排查

• 「Python 插件不加载」：检查 dsh.bundle 的 language: python 字段 + entry 是否能 import
• 「类型错误被静默吞掉」：工具函数抛异常时 SDK 默认转 ToolError，但模型看不到 stack trace；自定义错误时显式 raise ToolError("...", retryable=False)
• 「IPC 频繁卡顿」：合并多次小调用为一次大调用（批量 fetch / 批量 read），或改回 TS

FAQ

Python 插件能调用 TS 插件的 tool 吗？

能但不建议——跨语言 IPC 不可控，调试困难。建议 Python 插件只暴露自己的 tool，TS 侧组合调用。

pydantic v1 和 v2 都支持吗？

只支持 pydantic v2（性能 + 类型描述更现代）。