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%