from __future__ import annotations import uuid from dataclasses import dataclass, field from pathlib import Path from typing import Callable from .client import HarnessClient, HarnessConfig from .models import JsonObject, Notification @dataclass(slots=True) class DeepSeekHarnessConfig: """Configuration for launching the local DeepSeek Harness SDK runtime. The runtime inherits the caller's environment by default, so existing DEEPSEEK_API_KEY and DEEPSEEK_BASE_URL settings keep working. Use ``env`` to intentionally override or inject variables for a subprocess. """ provider: str = "deepseek-official" model: str = "deepseek-v4-flash" max_tokens: int | None = None cwd: str | None = None runtime_cwd: str | None = None session_root: str | None = None cordis: str | None = None env: dict[str, str] = field(default_factory=dict) runtime_bin: str | None = None launch_args_override: tuple[str, ...] | None = None request_timeout_seconds: float | None = None shutdown_timeout_seconds: float | None = 1.0 base_url: str | None = None api_key: str | None = None @dataclass(slots=True) class RunResult: session_id: str final_response: str events: list[JsonObject] notifications: list[Notification] session_root: str | None = None class DeepSeekHarness: """Reusable synchronous SDK for running DeepSeek Harness agent turns. The runtime subprocess starts lazily and remains owned by this instance across calls to :meth:`run`. Use the instance as a context manager, or call :meth:`close` explicitly when finished, so the subprocess is always reaped. """ def __init__(self, config: DeepSeekHarnessConfig | None = None, **kwargs: object) -> None: if config is not None and kwargs: raise TypeError("pass either DeepSeekHarnessConfig or keyword options, not both") self.config = config or DeepSeekHarnessConfig(**kwargs) cwd = str(Path(self.config.cwd or Path.cwd()).resolve()) runtime_cwd = str(Path(self.config.runtime_cwd).resolve()) if self.config.runtime_cwd is not None else cwd self._cwd = cwd env = dict(self.config.env) if self.config.session_root is not None: env["DSH_SESSION_ROOT"] = self.config.session_root if self.config.cordis is not None: env["DSH_CORDIS_CONFIG"] = self.config.cordis env["DSH_CWD"] = cwd if self.config.base_url is not None: env["DEEPSEEK_BASE_URL"] = self.config.base_url if self.config.api_key is not None: env["DEEPSEEK_API_KEY"] = self.config.api_key self._client = HarnessClient( HarnessConfig( runtime_bin=self.config.runtime_bin, launch_args_override=self.config.launch_args_override, cwd=runtime_cwd, env=env, request_timeout_seconds=self.config.request_timeout_seconds, shutdown_timeout_seconds=self.config.shutdown_timeout_seconds, ) ) self._initialized = False def __enter__(self) -> "DeepSeekHarness": self.start() return self def __exit__(self, _exc_type, _exc, _tb) -> None: self.close() @property def client(self) -> HarnessClient: return self._client def start(self) -> None: if self._initialized: return self._client.start() self._client.initialize( cwd=self._cwd, provider=self.config.provider, model=self.config.model, max_tokens=self.config.max_tokens, ) self._initialized = True def close(self) -> None: self._client.close() self._initialized = False def start_session(self, session_id: str | None = None) -> "Session": self.start() return Session(self, session_id or f"session-{uuid.uuid4().hex}") def run( self, input: str | list[JsonObject], *, session_id: str | None = None, on_notification: Callable[[Notification], None] | None = None, ) -> RunResult: return self.start_session(session_id).run(input, on_notification=on_notification) class Session: def __init__(self, harness: DeepSeekHarness, session_id: str) -> None: self.harness = harness self.id = session_id def run( self, input: str | list[JsonObject], *, on_notification: Callable[[Notification], None] | None = None, ) -> RunResult: content_blocks = normalize_input(input) notifications: list[Notification] = [] events: list[JsonObject] = [] def collect(notification: Notification) -> None: notifications.append(notification) if on_notification is not None: on_notification(notification) if ( notification.method == "session.event" and notification.payload.get("sessionId") == self.id ): event = notification.payload.get("event") if isinstance(event, dict): events.append(event) with self.harness.client.subscribe_session_notifications(self.id) as subscription: message_id = self.harness.client.session_prompt( self.id, content_blocks, notification_subscription=subscription, ) received = False while True: notification = subscription.next() if not received: if not _is_inbox_receipt(notification, self.id, message_id): continue received = True collect(notification) if ( notification.method == "session.status" and notification.payload.get("sessionId") == self.id and notification.payload.get("status") == "idle" ): break return RunResult( session_id=self.id, final_response=final_response(events), events=events, notifications=notifications, session_root=self.harness.config.session_root, ) def _is_inbox_receipt(notification: Notification, session_id: str, message_id: str) -> bool: if notification.method != "session.event" or notification.payload.get("sessionId") != session_id: return False event = notification.payload.get("event") if not isinstance(event, dict) or event.get("type") != "agent/inbox/spliced": return False data = event.get("data") inserted = data.get("inserted") if isinstance(data, dict) else None return isinstance(inserted, list) and any( isinstance(message, dict) and message.get("id") == message_id for message in inserted ) def normalize_input(input: str | list[JsonObject]) -> list[JsonObject]: if isinstance(input, str): return [{"type": "text", "text": input}] return input def final_response(events: list[JsonObject]) -> str: for event in reversed(events): if event.get("type") != "assistant/message": continue data = event.get("data") if not isinstance(data, dict): continue message = data.get("message") content_owner = message if isinstance(message, dict) else data content = content_owner.get("content") if not isinstance(content, list): continue parts: list[str] = [] for block in content: if isinstance(block, dict) and block.get("type") == "text": parts.append(str(block.get("text") or "")) return "".join(parts) return ""