DeepSeek Harness: Python SDK: الواجهة البرمجية

pip install deepseek-dsh

import dsh

print(dsh.__version__)
# 0.x.x

# تشغيل خادم DSH أولاً (وضع Headless)
npx @deepseek-ai/dsh headless --port 3080

npx @deepseek-ai/dsh headless --port 8080

from dsh import DSHClient

client = DSHClient(
    base_url="http://127.0.0.1:3080",
    api_key="your-api-key"  # اختياري، إذا كان DSH يملك مصادقة مُعدة
)

import os
from dsh import DSHClient

client = DSHClient.from_env()
# يقرأ متغيرات البيئة DSH_BASE_URL وDSH_API_KEY

session = client.create_session(
    workspace="/home/alice/my-project",
    model="deepseek-chat",
    mode="standard"
)

print(f"معرف الجلسة: {session.id}")
print(f"مساحة العمل: {session.workspace}")
print(f"النموذج: {session.model}")

session = client.create_session(
    workspace="/home/alice/my-project",
    model="deepseek-chat",
    mode="ptc",
    sandbox="permissive",
    settings={
        "temperature": 0.7,
        "max_tokens": 4096
    }
)

session = client.get_session("sess_abc123")
print(f"جلسة مستعادة: {session.id}")
print(f"الرسائل: {len(session.messages)}")

response = session.send("ساعدني في التحقق من package.json الخاص بالمشروع")

print(response.content)
# يُظهر package.json الخاص بالمشروع...

print(f"الأدوات المستخدمة: {len(response.tool_calls)}")
for tool in response.tool_calls:
    print(f"  - {tool.name}: {tool.status}")

class AgentResponse:
    content: str               # رد Agent النصي
    tool_calls: list[ToolCall] # سجلات استدعاءات الأدوات
    model: str                 # النموذج المستخدم
    tokens_used: int           # الرموز المستهلكة
    duration_ms: int           # وقت الاستجابة

class ToolCall:
    name: str                  # اسم الأداة
    params: dict               # معاملات الاستدعاء
    status: str                # حالة التنفيذ
    result: Any                # نتيجة التنفيذ
    duration_ms: int           # وقت التنفيذ

# الدور الأول
resp1 = session.send("اعرض محتويات src/app.ts")
print(resp1.content)

# الدور الثاني (السياق تلقائي)
resp2 = session.send("أضف وسيط معالجة أخطاء لهذا الملف")
print(resp2.content)

# الدور الثالث
resp3 = session.send("شغّل الاختبارات للتأكد من عدم وجود أخطاء")
print(resp3.content)

response = session.send("أنشئ src/utils/helpers.ts، اكتب دالة debounce")

for tool in response.tool_calls:
    print(f"الأداة: {tool.name}")
    print(f"المعاملات: {tool.params}")
    print(f"النتيجة: {tool.result}")

def on_approval(tool_name: str, params: dict) -> bool:
    print(f"موافقة مطلوبة: {tool_name}")
    print(f"المعاملات: {params}")
    
    # سماح تلقائي للعمليات الآمنة
    if tool_name == "file_edit" and params.get("action") == "read":
        return True
    
    # العمليات الأخرى تتطلب تأكيدًا يدويًا
    confirm = input(f"هل تسمح بـ {tool_name}؟ (y/n): ")
    return confirm.lower() == "y"

session = client.create_session(
    workspace="/home/alice/project",
    approval_callback=on_approval
)

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

response = session.send("حلل تغطية الاختبار في المشروع")

for tool in response.tool_calls:
    if tool.name == "shell":
        output = tool.result.get("stdout", "")
        if "Coverage" in output:
            print(f"تغطية الاختبار: {output}")
    elif tool.name == "search":
        files = tool.result.get("files", [])
        print(f"وُجد {len(files)} ملف اختبار")
    elif tool.name == "file_edit":
        action = tool.params.get("action")
        path = tool.params.get("path")
        print(f"ملف {action}: {path}")

for chunk in session.send_stream("اشرح تصميم بنية هذا المشروع بالتفصيل"):
    if chunk.type == "content":
        print(chunk.text, end="", flush=True)
    elif chunk.type == "tool_call":
        print(f"\n[أداة: {chunk.tool_name}]")
    elif chunk.type == "tool_result":
        print(f"[تم استلام نتيجة الأداة]")

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[موافقة تلقائية: {chunk.tool_name}]")

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"الخطة: {response.content}")
print(f"الأدوات المستخدمة: {len(response.tool_calls)}")
print(f"الرموز: {response.tokens_used}")

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']} استدعاءات أدوات")

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