DeepSeek Harness: Python SDK
最后更新：2026-08-31
Web UI 和 CLI 适合人类交互，但当你需要将 Agent 集成到自动化管线、CI/CD 流程或自定义应用中时，Python SDK 就是你的入口——几行代码即可启动 Agent 会话，获取结构化响应。
💡 提示：Python SDK 适合需要程序化控制 Agent 的场景——批量处理、自动化测试、数据管线。如果你只是日常使用，Web UI 和 CLI 更方便。
📋 前置知识：已完成 06-tools.md，了解工具系统；有 Python 基础
1. 你将学到
- Python SDK 安装与初始化
- 创建会话与发送消息
- 获取 Agent 响应与工具调用结果
- 流式输出处理
- 工具调用拦截与自定义
- 错误处理与超时管理
2. 安装与初始化
(1) 安装 SDK
BASH
pip install deepseek-dsh
验证安装：
PYTHON
import dsh
print(dsh.__version__)
# 0.x.x
(2) 前置条件
Python SDK 需要 DSH 服务端运行中：
BASH
# 先启动 DSH 服务端（Headless 模式）
npx @deepseek-ai/dsh headless --port 3080
或者指定自定义端口：
BASH
npx @deepseek-ai/dsh headless --port 8080
(3) 初始化客户端
▶ 示例 1创建 SDK 客户端
PYTHON
from dsh import DSHClient
client = DSHClient(
base_url="http://127.0.0.1:3080",
api_key="your-api-key" # 可选，如果 DSH 配置了认证
)
输出：
TEXT 📖 仅展示✅ DSHClient initialized 🔗 Base URL: http://127.0.0.1:3080 🔑 API Key: configured
▶ 示例 2使用环境变量初始化
PYTHON
import os
from dsh import DSHClient
client = DSHClient.from_env()
# 读取 DSH_BASE_URL 和 DSH_API_KEY 环境变量
输出：
TEXT 📖 仅展示✅ DSHClient initialized from environment 🔗 DSH_BASE_URL: http://127.0.0.1:3080 🔑 DSH_API_KEY: (set)
3. 创建会话
(1) 创建新会话
▶ 示例 3创建会话
PYTHON
session = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="standard"
)
print(f"Session ID: {session.id}")
print(f"Workspace: {session.workspace}")
print(f"Model: {session.model}")
输出：
TEXT 📖 仅展示Session ID: sess_abc123 Workspace: /home/alice/my-project Model: deepseek-chat
(2) 会话配置
PYTHON
session = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="ptc",
sandbox="permissive",
settings={
"temperature": 0.7,
"max_tokens": 4096
}
)
(3) 恢复已有会话
▶ 示例 4通过 Session ID 恢复
PYTHON
session = client.get_session("sess_abc123")
print(f"Restored session: {session.id}")
print(f"Messages: {len(session.messages)}")
输出：
TEXT 📖 仅展示Restored session: sess_abc123 Messages: 12
4. 发送消息与获取响应
(1) 基本消息发送
▶ 示例 5发送消息并获取完整响应
PYTHON
response = session.send("帮我查看项目的 package.json")
print(response.content)
# 这个项目的 package.json 显示...
print(f"Tools used: {len(response.tool_calls)}")
for tool in response.tool_calls:
print(f" - {tool.name}: {tool.status}")
交互流程：
🤖 Agent: 正在查看 package.json... 🔧 Using tool: file_edit (read) → Path: package.json 🤖 Agent: 这个项目的 package.json 显示... 📊 Tools used: 1 - file_edit: success⚠️ 你的实际输出取决于项目内容和模型，但工具调用流程应相似。
(2) 响应结构
PYTHON
class AgentResponse:
content: str # Agent 的文本回复
tool_calls: list[ToolCall] # 工具调用记录
model: str # 使用的模型
tokens_used: int # 消耗的 token 数
duration_ms: int # 响应耗时
class ToolCall:
name: str # 工具名称
params: dict # 调用参数
status: str # 执行状态
result: Any # 执行结果
duration_ms: int # 执行耗时
(3) 带上下文的多轮对话
▶ 示例 6多轮对话
PYTHON
# 第一轮
resp1 = session.send("查看 src/app.ts 的内容")
print(resp1.content)
# 第二轮（自动带上下文）
resp2 = session.send("给这个文件添加错误处理中间件")
print(resp2.content)
# 第三轮
resp3 = session.send("运行测试确保没有破坏现有功能")
print(resp3.content)
交互流程：
[第1轮] 🤖 Agent: 正在读取 src/app.ts... 🔧 Using tool: file_edit (read) → src/app.ts [第2轮] 🤖 Agent: 正在添加错误处理中间件... 🔧 Using tool: file_edit (edit) → src/app.ts ⚠️ Approval: edit src/app.ts → ✅ Allowed [第3轮] 🤖 Agent: 正在运行测试... 🔧 Using tool: shell → npm test ✅ 所有测试通过⚠️ 多轮对话中每轮的工具调用和输出都因模型而异，但上下文会自动累积。
5. 工具调用
(1) 自动工具调用
标准模式下，Agent 自动决定何时调用工具：
PYTHON
response = session.send("创建 src/utils/helpers.ts，写一个 debounce 函数")
for tool in response.tool_calls:
print(f"Tool: {tool.name}")
print(f"Params: {tool.params}")
print(f"Result: {tool.result}")
(2) 工具审批处理
当 Agent 的操作需要审批时，SDK 提供回调机制：
▶ 示例 7审批回调
PYTHON
def on_approval(tool_name: str, params: dict) -> bool:
print(f"Approval requested: {tool_name}")
print(f"Params: {params}")
# 自动允许安全操作
if tool_name == "file_edit" and params.get("action") == "read":
return True
# 其他操作需要人工确认
confirm = input(f"Allow {tool_name}? (y/n): ")
return confirm.lower() == "y"
session = client.create_session(
workspace="/home/alice/project",
approval_callback=on_approval
)
交互流程：
🤖 Agent: 正在读取配置文件... ⚠️ Approval requested: file_edit Params: {action: "read", path: "src/config.ts"} → auto-approved (read action) 🤖 Agent: 配置文件内容如下...⚠️ 审批回调的触发取决于 Agent 的工具选择，实际操作会不同。
(3) 禁用特定工具
PYTHON
session = client.create_session(
workspace="/home/alice/project",
disabled_tools=["shell", "sandbox"]
)
(4) 工具调用结果处理
▶ 示例 8详细处理工具结果
PYTHON
response = session.send("分析项目的测试覆盖率")
for tool in response.tool_calls:
if tool.name == "shell":
output = tool.result.get("stdout", "")
if "Coverage" in output:
print(f"Test coverage: {output}")
elif tool.name == "search":
files = tool.result.get("files", [])
print(f"Found {len(files)} test files")
elif tool.name == "file_edit":
action = tool.params.get("action")
path = tool.params.get("path")
print(f"File {action}: {path}")
交互流程：
🤖 Agent: 正在分析测试覆盖率... 🔧 Using tool: shell → npm run test:coverage 📊 Test coverage: 78% statements, 65% branches 🔧 Using tool: search → 查找测试文件 📊 Found 12 test files 🔧 Using tool: file_edit (read) → 读取未覆盖模块 🤖 Agent: 覆盖率分析完成...⚠️ 你的实际工具调用序列和输出取决于项目，但结果处理流程应相似。
6. 流式输出处理
(1) 启用流式输出
对于长时间响应，使用流式输出实时获取结果：
▶ 示例 9流式输出
PYTHON
for chunk in session.send_stream("详细解释这个项目的架构设计"):
if chunk.type == "content":
print(chunk.text, end="", flush=True)
elif chunk.type == "tool_call":
print(f"\n[Tool: {chunk.tool_name}]")
elif chunk.type == "tool_result":
print(f"[Tool result received]")
交互流程：
[content] 这个项目采用了分层架构... [content] 主要分为三个模块... [tool_call] Tool: search [tool_result] Found 8 files [content] 根据分析，架构设计如下... [done] Tokens: 2450, Duration: 3200ms⚠️ 流式输出的具体文本和工具调用完全取决于模型，但事件类型序列固定。
(2) 流式输出的事件类型
| 事件类型 | 说明 | 数据字段 |
|---|---|---|
content |
文本内容片段 | text |
tool_call |
工具调用开始 | tool_name, params |
tool_result |
工具执行结果 | tool_name, result |
approval |
审批请求 | tool_name, params |
done |
响应完成 | tokens_used, duration_ms |
error |
错误发生 | code, message |
(3) 流式 + 审批结合
▶ 示例 10流式输出中处理审批
PYTHON
def auto_approve(tool_name: str, params: dict) -> bool:
safe_actions = ["read", "search"]
if params.get("action") in safe_actions:
return True
return False
for chunk in session.send_stream(
"重构所有控制器，添加错误处理",
approval_callback=auto_approve
):
if chunk.type == "content":
print(chunk.text, end="")
elif chunk.type == "approval":
print(f"\n[Auto-approved: {chunk.tool_name}]")
交互流程：
[content] 正在重构所有控制器... [tool_call] Tool: file_edit (read) [Auto-approved: file_edit] [tool_call] Tool: file_edit (edit) [content] 已完成 3/8 个控制器... [tool_call] Tool: file_edit (edit) [content] 重构完成，共修改 8 个文件⚠️ 自动审批只允许安全操作（read/search），edit 仍需人工确认（若未配置）。
7. 高级功能
(1) Headless 模式集成
SDK 最常见的场景是与 Headless 模式配合，实现无人值守的 Agent 执行：
▶ 示例 11完整的 Headless 工作流
PYTHON
from dsh import DSHClient
client = DSHClient(base_url="http://127.0.0.1:3080")
def auto_approve(tool_name: str, params: dict) -> bool:
safe_tools = ["search", "file_edit", "plan"]
if tool_name in safe_tools:
action = params.get("action", "")
if action in ["read", "create"]:
return True
return False
session = client.create_session(
workspace="/home/alice/project",
model="deepseek-coder",
mode="ptc",
approval_callback=auto_approve
)
response = session.send(
"为所有路由添加输入验证中间件，确保请求参数符合预期类型"
)
print(f"Plan: {response.content}")
print(f"Tools used: {len(response.tool_calls)}")
print(f"Tokens: {response.tokens_used}")
交互流程：
🤖 Agent: [PTC 模式] 正在制定计划... 📋 Plan: 为所有路由添加输入验证中间件 1. 搜索路由文件 → search 2. 读取路由文件 → file_edit (read) 3. 创建中间件 → file_edit (create) 4. 修改路由 → file_edit (edit) 🔧 Executing plan... ✅ Plan completed: 4 steps, 6 tool calls 📊 Tokens used: 8450⚠️ Headless 工作流的计划内容和工具调用完全取决于任务，但 PTC 流程固定。
(2) 并发会话
▶ 示例 12多会话并行
PYTHON
import concurrent.futures
def process_file(filepath: str):
client = DSHClient(base_url="http://127.0.0.1:3080")
session = client.create_session(workspace="/home/alice/project")
response = session.send(f"为 {filepath} 添加单元测试")
return {"file": filepath, "tests_added": len(response.tool_calls)}
files = [
"src/utils/format.ts",
"src/utils/validate.ts",
"src/routes/users.ts"
]
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(process_file, files))
for r in results:
print(f"{r['file']}: {r['tests_added']} tool calls")
交互流程：
[Session 1] 🤖 正在为 src/utils/format.ts 添加测试... [Session 2] 🤖 正在为 src/utils/validate.ts 添加测试... [Session 3] 🤖 正在为 src/routes/users.ts 添加测试... [Session 1] ✅ 完成, 3 tool calls [Session 2] ✅ 完成, 2 tool calls [Session 3] ✅ 完成, 4 tool calls 📊 src/utils/format.ts: 3 tool calls 📊 src/utils/validate.ts: 2 tool calls 📊 src/routes/users.ts: 4 tool calls⚠️ 并发会话的执行顺序不确定，每个 Agent 的工具调用也不同。
(3) 错误处理
▶ 示例 13错误处理
PYTHON
from dsh import DSHClient, DSHTimeoutError, DSHConnectionError
client = DSHClient(base_url="http://127.0.0.1:3080")
try:
session = client.create_session(workspace="/home/alice/project")
response = session.send("帮我修复所有 TypeScript 错误", timeout=300)
except DSHTimeoutError:
print("Agent 响应超时，请简化任务或增加超时时间")
except DSHConnectionError:
print("无法连接 DSH 服务端，请检查是否已启动")
except Exception as e:
print(f"未知错误: {e}")
输出：
TEXT 📖 仅展示# 正常情况： ✅ Agent 响应完成 # 超时情况： Agent 响应超时，请简化任务或增加超时时间 # 连接失败： 无法连接 DSH 服务端，请检查是否已启动
8. SDK 与 Web UI/CLI 对比
| 维度 | Web UI | CLI | Python SDK |
|---|---|---|---|
| 交互方式 | 浏览器 | 终端 | 代码 |
| 适合用户 | 所有人 | 开发者 | 自动化工程师 |
| 审批机制 | 弹窗交互 | 命令行确认 | 回调函数 |
| 流式输出 | 实时渲染 | 终端输出 | 事件流 |
| 并发 | 单会话 | 单会话 | 多会话 |
| 集成能力 | 低 | 中 | 高 |
| 学习曲线 | 最低 | 低 | 中等 |
❓ 常见问题
Q SDK 需要单独安装 DSH 吗？
A 需要。SDK 是客户端，DSH 服务端仍需通过 npx 或源码启动。SDK 通过 HTTP API 与服务端通信。
Q Python SDK 支持 Python 2 吗？
A 不支持。Python SDK 需要 Python 3.8+。
Q SDK 调用是否加密？
A 本地通信默认不加密（http://）。生产环境建议配置 HTTPS 或通过 SSH 隧道访问。
Q 可以用 SDK 控制 CLI 模式的 Agent 吗？
A 不可以。SDK 对接的是 DSH 的 HTTP API（Headless 模式）。CLI 模式是独立的终端交互。
Q 流式输出和普通输出的结果一样吗？
A 是的，最终结果完全一致。流式只是实时返回内容片段，普通模式等待完整响应后一次性返回。
Q 如何调试 SDK 调用？
A 启用 debug 日志：
client = DSHClient(base_url="...", debug=True)。所有 HTTP 请求和响应会打印到控制台。 ---📖 小节
- Python SDK 通过
pip install @deepseek-ai/dsh-python安装 - SDK 需要 DSH 服务端（Headless 模式）运行中
- 创建会话 → 发送消息 → 获取响应是核心三步
- 工具审批通过回调函数处理
- 流式输出
send_stream()适合长时间响应 - 支持并发会话、错误处理、超时控制
- SDK 适合自动化集成，日常使用推荐 Web UI
📝 作业
1. ⭐ 基础题：安装 Python SDK，启动 DSH Headless 模式，用 SDK 创建一个会话并发送"Hello"消息，打印 Agent 的回复内容。
2. ⭐⭐ 进阶题：编写一个 Python 脚本，用 SDK 让 Agent 读取项目的 README.md 并生成一份项目概要报告，保存到 project-summary.txt。
3. ⭐⭐⭐ 挑战题：编写一个批量处理脚本，用并发会话同时让 3 个 Agent 分析项目中不同目录的代码质量，汇总输出一份完整的代码质量报告。
