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. 学習内容

SDK 呼び出しフロー


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"  # Optional, if DSH has authentication configured
)

▶ サンプル 2:環境変数を使った初期化

PYTHON
import os
from dsh import DSHClient

client = DSHClient.from_env()
# Reads DSH_BASE_URL and DSH_API_KEY environment variables

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}")

(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:セッション ID による復元

PYTHON
session = client.get_session("sess_abc123")
print(f"Restored session: {session.id}")
print(f"Messages: {len(session.messages)}")

4. メッセージの送信と応答の取得

(1) 基本的なメッセージ送信

▶ サンプル 5:メッセージの送信と完全な応答の取得

PYTHON
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) 応答構造

PYTHON
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:マルチターン会話

PYTHON
# 第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 がツールを呼び出すタイミングを自動的に決定します:

PYTHON
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:承認コールバック

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
)

(3) 特定ツールの無効化

PYTHON
session = client.create_session(
    workspace="/home/alice/project",
    disabled_tools=["shell", "sandbox"]
)

(4) ツール呼び出し結果の処理

▶ サンプル 8:詳細なツール結果処理

PYTHON
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:ストリーミング出力

PYTHON
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:ストリーミング出力での承認処理

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(
    "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 ワークフロー

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(
    "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:並列マルチセッション

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"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:エラー処理

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("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
インタラクション方法 ブラウザ ターミナル コード
適している対象 すべての人 開発者 自動化エンジニア
承認機構 ポップアップインタラクション コマンドライン確認 コールバック関数
ストリーミング出力 リアルタイムレンダリング ターミナル出力 イベントストリーム
並行性 シングルセッション シングルセッション マルチセッション
統合能力 低 中 高
学習曲線 最も低い 低 中

❓ よくある質問

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 デバッグログを有効にしてください:client = DSHClient(base_url="...", debug=True)。すべての HTTP リクエストとレスポンスがコンソールに出力されます。

📖 まとめ


📝 練習問題

1. ⭐ 基礎:Python SDK をインストールし、Headless モードで DSH を起動し、SDK を使ってセッションを作成して「Hello」メッセージを送信し、Agent の返信内容を表示してください。

2. ⭐⭐ 応用:SDK を使って Agent にプロジェクトの README.md を読み取らせ、プロジェクト概要レポートを生成して project-summary.txt に保存する Python スクリプトを書いてください。

3. ⭐⭐⭐ チャレンジ:並行セッションを使って3つの Agent が同時に異なるプロジェクトディレクトリのコード品質を分析し、統合コード品質レポートを出力するバッチ処理スクリプトを書いてください。

Web-Tutorial.com

Web-Tutorial 技術チーム

複数の開発者によって共同維持されているプログラミングチュートリアルプラットフォーム。各チュートリアルは専門分野の開発者が執筆・レビューしています。正確で信頼性の高いコンテンツを目指しています — 問題を見つけた場合はお知らせください。

100%