DeepSeek Harness: Python SDK 入門
最終更新:2026-08-31
Web UI と CLI は人間のインタラクションに最適ですが、Agent を自動化パイプライン、CI/CD ワークフロー、カスタムアプリケーションに統合する必要がある場合、Python SDK がエントリポイントになります——数行のコードで Agent セッションを開始し、構造化された応答を取得できます。
📋 前提知識:06-tools.md の完了、ツールシステムに精通していること;基本的な Python 知識
1. 学習内容
- Python SDK のインストールと初期化
- セッションの作成とメッセージの送信
- Agent の応答とツール呼び出し結果の取得
- ストリーミング出力の処理
- ツール呼び出しのインターセプトとカスタマイズ
- エラー処理とタイムアウト管理
2. インストールと初期化
(1) SDK のインストール
pip install deepseek-dsh
インストールの確認:
import dsh
print(dsh.__version__)
# 0.x.x
(2) 前提条件
Python SDK は DSH サーバーが実行されている必要があります:
# 先に DSH サーバーを起動(Headless モード)
npx @deepseek-ai/dsh headless --port 3080
またはカスタムポートを指定:
npx @deepseek-ai/dsh headless --port 8080
(3) クライアントの初期化
▶ サンプル 1:SDK クライアントの作成
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
)
▶ サンプル 2:環境変数を使った初期化
import os
from dsh import DSHClient
client = DSHClient.from_env()
# Reads DSH_BASE_URL and DSH_API_KEY environment variables
3. セッションの作成
(1) 新しいセッションの作成
▶ サンプル 3:セッションの作成
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 = client.create_session(
workspace="/home/alice/my-project",
model="deepseek-chat",
mode="ptc",
sandbox="permissive",
settings={
"temperature": 0.7,
"max_tokens": 4096
}
)
(3) 既存のセッションの復元
▶ サンプル 4:セッション ID による復元
session = client.get_session("sess_abc123")
print(f"Restored session: {session.id}")
print(f"Messages: {len(session.messages)}")
4. メッセージの送信と応答の取得
(1) 基本的なメッセージ送信
▶ サンプル 5:メッセージの送信と完全な応答の取得
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) 応答構造
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) コンテキスト付きマルチターン会話
▶ サンプル 6:マルチターン会話
# 第1ターン
resp1 = session.send("View the contents of src/app.ts")
print(resp1.content)
# 第2ターン(コンテキストは自動的に維持)
resp2 = session.send("Add error handling middleware to this file")
print(resp2.content)
# 第3ターン
resp3 = session.send("Run tests to make sure nothing is broken")
print(resp3.content)
5. ツール呼び出し
(1) 自動ツール呼び出し
Standard モードでは、Agent がツールを呼び出すタイミングを自動的に決定します:
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) ツール承認の処理
Agent の操作に承認が必要な場合、SDK はコールバック機構を提供します:
▶ サンプル 7:承認コールバック
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
)
(3) 特定ツールの無効化
session = client.create_session(
workspace="/home/alice/project",
disabled_tools=["shell", "sandbox"]
)
(4) ツール呼び出し結果の処理
▶ サンプル 8:詳細なツール結果処理
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. ストリーミング出力の処理
(1) ストリーミング出力の有効化
長い応答には、ストリーミング出力でリアルタイムに結果を取得できます:
▶ サンプル 9:ストリーミング出力
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) ストリーミング出力イベントタイプ
| イベントタイプ | 説明 | データフィールド |
|---|---|---|
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:ストリーミング出力での承認処理
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. 高度な機能
(1) Headless モード統合
SDK の最も一般的なユースケースは、Headless モードと組み合わせた無人 Agent 実行です:
▶ サンプル 11:完全な Headless ワークフロー
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) 並行セッション
▶ サンプル 12:並列マルチセッション
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) エラー処理
▶ サンプル 13:エラー処理
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 の比較
| 次元 | Web UI | CLI | Python SDK |
|---|---|---|---|
| インタラクション方法 | ブラウザ | ターミナル | コード |
| 適している対象 | すべての人 | 開発者 | 自動化エンジニア |
| 承認機構 | ポップアップインタラクション | コマンドライン確認 | コールバック関数 |
| ストリーミング出力 | リアルタイムレンダリング | ターミナル出力 | イベントストリーム |
| 並行性 | シングルセッション | シングルセッション | マルチセッション |
| 統合能力 | 低 | 中 | 高 |
| 学習曲線 | 最も低い | 低 | 中 |
❓ よくある質問
client = DSHClient(base_url="...", debug=True)。すべての HTTP リクエストとレスポンスがコンソールに出力されます。📖 まとめ
- Python SDK は
pip install deepseek-dshでインストール - SDK は DSH サーバー(Headless モード)の実行が必要
- セッション作成 → メッセージ送信 → 応答取得がコアの3ステップ
- ツール承認はコールバック関数で処理
- ストリーミング出力
send_stream()は長い応答に適している - 並行セッション、エラー処理、タイムアウト制御をサポート
- SDK は自動化統合向け;日常使用には Web UI を推奨
📝 練習問題
1. ⭐ 基礎:Python SDK をインストールし、Headless モードで DSH を起動し、SDK を使ってセッションを作成して「Hello」メッセージを送信し、Agent の返信内容を表示してください。
2. ⭐⭐ 応用:SDK を使って Agent にプロジェクトの README.md を読み取らせ、プロジェクト概要レポートを生成して project-summary.txt に保存する Python スクリプトを書いてください。
3. ⭐⭐⭐ チャレンジ:並行セッションを使って3つの Agent が同時に異なるプロジェクトディレクトリのコード品質を分析し、統合コード品質レポートを出力するバッチ処理スクリプトを書いてください。