DeepSeek Harness: Python SDK
Last updated: 2026-08-31
The Web UI and CLI are great for human interaction, but when you need to integrate the Agent into automation pipelines, CI/CD workflows, or custom applications, the Python SDK is your entry point — a few lines of code to start an Agent session and get structured responses.
📋 Prerequisites: Completed 06-tools.md, familiar with the tool system; basic Python knowledge
1. What You'll Learn
- Python SDK installation and initialization
- Creating sessions and sending messages
- Getting Agent responses and tool call results
- Streaming output handling
- Tool call interception and customization
- Error handling and timeout management
2. Installation and Initialization
(1) Installing the SDK
pip install deepseek-dsh
Verify installation:
import dsh
print(dsh.__version__)
# 0.x.x
(2) Prerequisites
The Python SDK requires the DSH server to be running:
# Start DSH server first (Headless mode)
npx @deepseek-ai/dsh headless --port 3080
Or specify a custom port:
npx @deepseek-ai/dsh headless --port 8080
(3) Initialize the Client
▶ Example 1: Creating an SDK Client
from dsh import DSHClient
client = DSHClient(
base_url="http://127.0.0.1:3080",
api_key="your-api-key" # Optional, if DSH has authentication configured
)
▶ Example 2: Initialize Using Environment Variables
import os
from dsh import DSHClient
client = DSHClient.from_env()
# Reads DSH_BASE_URL and DSH_API_KEY environment variables
3. Creating a Session
(1) Creating a New Session
▶ Example 3: Creating a Session
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}")
(2) Session Configuration
session = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="ptc",
sandbox="permissive",
settings={
"temperature": 0.7,
"max_tokens": 4096
}
)
(3) Restoring an Existing Session
▶ Example 4: Restore by Session ID
session = client.get_session("sess_abc123")
print(f"Restored session: {session.id}")
print(f"Messages: {len(session.messages)}")
4. Sending Messages and Getting Responses
(1) Basic Message Sending
▶ Example 5: Send a Message and Get a Complete Response
response = session.send("Help me check the project's package.json")
print(response.content)
# The project's package.json shows...
print(f"Tools used: {len(response.tool_calls)}")
for tool in response.tool_calls:
print(f" - {tool.name}: {tool.status}")
(2) Response Structure
class AgentResponse:
content: str # Agent's text reply
tool_calls: list[ToolCall] # Tool call records
model: str # Model used
tokens_used: int # Tokens consumed
duration_ms: int # Response time
class ToolCall:
name: str # Tool name
params: dict # Call parameters
status: str # Execution status
result: Any # Execution result
duration_ms: int # Execution time
(3) Multi-turn Conversations with Context
▶ Example 6: Multi-turn Conversation
# First turn
resp1 = session.send("View the contents of src/app.ts")
print(resp1.content)
# Second turn (context is automatic)
resp2 = session.send("Add error handling middleware to this file")
print(resp2.content)
# Third turn
resp3 = session.send("Run tests to make sure nothing is broken")
print(resp3.content)
5. Tool Calls
(1) Automatic Tool Calls
In Standard mode, the Agent decides when to call tools automatically:
response = session.send("Create src/utils/helpers.ts, write a debounce function")
for tool in response.tool_calls:
print(f"Tool: {tool.name}")
print(f"Params: {tool.params}")
print(f"Result: {tool.result}")
(2) Tool Approval Handling
When an Agent's operation requires approval, the SDK provides a callback mechanism:
▶ Example 7: Approval Callback
def on_approval(tool_name: str, params: dict) -> bool:
print(f"Approval requested: {tool_name}")
print(f"Params: {params}")
# Auto-allow safe operations
if tool_name == "file_edit" and params.get("action") == "read":
return True
# Other operations require manual confirmation
confirm = input(f"Allow {tool_name}? (y/n): ")
return confirm.lower() == "y"
session = client.create_session(
workspace="/home/alice/project",
approval_callback=on_approval
)
(3) Disabling Specific Tools
session = client.create_session(
workspace="/home/alice/project",
disabled_tools=["shell", "sandbox"]
)
(4) Tool Call Result Handling
▶ Example 8: Detailed Tool Result Processing
response = session.send("Analyze the project's test coverage")
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}")
6. Streaming Output Handling
(1) Enabling Streaming Output
For long responses, use streaming output to get results in real-time:
▶ Example 9: Streaming Output
for chunk in session.send_stream("Explain this project's architecture design in detail"):
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]")
(2) Streaming Output Event Types
| Event Type | Description | Data Fields |
|---|---|---|
content |
Text content fragment | text |
tool_call |
Tool call started | tool_name, params |
tool_result |
Tool execution result | tool_name, result |
approval |
Approval request | tool_name, params |
done |
Response complete | tokens_used, duration_ms |
error |
Error occurred | code, message |
(3) Combining Streaming with Approval
▶ Example 10: Handling Approval in Streaming Output
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(
"Refactor all controllers, add error handling",
approval_callback=auto_approve
):
if chunk.type == "content":
print(chunk.text, end="")
elif chunk.type == "approval":
print(f"\n[Auto-approved: {chunk.tool_name}]")
7. Advanced Features
(1) Headless Mode Integration
The most common SDK use case is pairing with Headless mode for unattended Agent execution:
▶ Example 11: Complete Headless Workflow
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(
"Add input validation middleware for all routes, ensuring request parameters match expected types"
)
print(f"Plan: {response.content}")
print(f"Tools used: {len(response.tool_calls)}")
print(f"Tokens: {response.tokens_used}")
(2) Concurrent Sessions
▶ Example 12: Parallel Multi-Session
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"Add unit tests for {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")
(3) Error Handling
▶ Example 13: Error Handling
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("Help me fix all TypeScript errors", timeout=300)
except DSHTimeoutError:
print("Agent response timed out. Please simplify the task or increase the timeout")
except DSHConnectionError:
print("Cannot connect to DSH server. Please check if it's running")
except Exception as e:
print(f"Unknown error: {e}")
8. SDK vs. Web UI/CLI Comparison
| Dimension | Web UI | CLI | Python SDK |
|---|---|---|---|
| Interaction method | Browser | Terminal | Code |
| Suitable for | Everyone | Developers | Automation engineers |
| Approval mechanism | Popup interaction | Command-line confirmation | Callback function |
| Streaming output | Real-time rendering | Terminal output | Event stream |
| Concurrency | Single session | Single session | Multi-session |
| Integration capability | Low | Medium | High |
| Learning curve | Lowest | Low | Medium |
❓ FAQ
client = DSHClient(base_url="...", debug=True). All HTTP requests and responses will be printed to the console.📖 Summary
- Python SDK installs via
pip install @deepseek-ai/dsh-python - SDK requires DSH server (Headless mode) to be running
- Create session → Send message → Get response is the core three-step flow
- Tool approval is handled via callback functions
- Streaming output
send_stream()is suitable for long responses - Supports concurrent sessions, error handling, and timeout control
- SDK is suitable for automation integration; daily use recommends Web UI
📝 Exercises
1. ⭐ Basic: Install the Python SDK, start DSH in Headless mode, create a session using the SDK and send a "Hello" message, print the Agent's reply content.
2. ⭐⭐ Intermediate: Write a Python script that uses the SDK to have the Agent read the project's README.md and generate a project summary report, saving it to project-summary.txt.
3. ⭐⭐⭐ Challenge: Write a batch processing script that uses concurrent sessions to have 3 Agents simultaneously analyze code quality in different project directories, producing a consolidated code quality report.