Usage · 10

Python SDK in Practice: Connect a Client and Run Session Tasks

Install the paired runtime, configure workspaces and sessions, interpret activity intervals and notifications, and close safely.

Reading time
18 minutes
Sources verified

Verify the platform and runtime pairing

The published SDK requires Python 3.10 or newer and supports Linux x64, Linux arm64, and macOS 14 or newer on arm64. The minimal persistent-PTY example needs a POSIX terminal substrate and is not a Windows agent interface. Confirm architecture as well as operating-system version before debugging imports or subprocess startup.

Installing deepseek-harness-sdk also installs the exact same-version deepseek-harness-runtime-bin platform wheel. DeepSeekHarness launches that bundled single-file runtime by default, so normal users need no system Node.js and should not point the SDK at a separately installed dsh. SDK/runtime version pairing is part of the protocol contract.

bash
python --version
python -c "import platform; print(platform.system(), platform.machine())"
python -m pip show deepseek-harness-sdk deepseek-harness-runtime-bin

Install in isolation and run the checked-in example

Clone the repository for its example composition, create a virtual environment, and install the distribution from PyPI. Provide the DeepSeek credential through the process environment or a controlled secret injector. DEEPSEEK_BASE_URL selects a compatible proxy when the default endpoint is not appropriate; never embed a key in source or command arguments.

bash
git clone https://github.com/deepseek-ai/deepseek-harness.git
cd deepseek-harness
python -m venv .venv
. .venv/bin/activate
python -m pip install deepseek-harness-sdk
export DEEPSEEK_API_KEY='<your-api-key>'
bash
python examples/jsonrpc-agent/minimal.py \
  --workspace /absolute/path/to/disposable-workspace \
  --session-root /absolute/path/to/private-sessions \
  --session-id example-001 \
  "Inspect the repository and report the failing tests."

Success prints the final assistant response and creates a JSONL session log containing assembled requests and tool calls. Verify both rather than trusting prose alone. Keep the session root private because logs can contain prompts, paths, arguments, results, and model output.

Use DeepSeekHarness and Session deliberately

DeepSeekHarness is the process owner. It starts the runtime lazily on first use and reuses it across calls until close. start_session(session_id) returns a Session handle; harness.run(input, session_id=...) is the convenience path. Reusing the same harness and session id continues durable conversation and the session-owned persistent Bash process, including working directory, exported variables, and shell functions.

python
from deepseek_harness import DeepSeekHarness

with DeepSeekHarness(
    provider='deepseek-official',
    model='deepseek-v4-flash',
    max_tokens=49_152,
    cwd='/work/repo',
    session_root='/work/sessions',
    cordis='examples/jsonrpc-agent/minimal.cordis.yml',
) as harness:
    session = harness.start_session('review-001')
    first = session.run('Inspect failing tests without editing.')
    second = session.run('Now fix only the confirmed defect.')

Use a fresh session id for an independent job, evaluation repeat, or different trust boundary. Reuse an id only when continuation and persistent shell state are intentional. If no id is supplied, the high-level API generates one; record result.session_id rather than guessing it.

Separate cwd, runtime_cwd, session_root, and Cordis config

cwd selects the agent workspace and is resolved to an absolute path before launch and handshake. runtime_cwd controls the subprocess working directory when that differs. session_root is a convenience that sets DSH_SESSION_ROOT for persistence; it should live outside disposable source when logs must survive checkout cleanup, but inside protected storage. cordis points to the deployment composition and must retain the SDK JSON-RPC server entry.

python
from pathlib import Path
workspace = Path('/tmp/eval/worktree').resolve()
sessions = Path('/var/private/dsh-sessions').resolve()
config = Path('examples/jsonrpc-agent/minimal.cordis.yml').resolve()

assert workspace.is_dir()
assert config.is_file()
# Pass str(workspace), str(sessions), and str(config) to the harness.

Provider, model, and max_tokens are applied request options. Deployment persona, tool set, persistence backend, sandbox, and compaction belong in Cordis composition. The bundled default config is injected only when using the bundled runtime without an explicit cordis or non-empty DSH_CORDIS_CONFIG. Explicit runtime_bin, bridge_bin, or launch_args_override also disables that automatic injection.

Interpret RunResult as an activity interval

Session.run() queues the prompt, waits until its MessageId appears in a durable inbox receipt, then collects through the next whole-agent idle. RunResult contains session_id, final_response, finish_reason, events, notifications, and session_root. final_response is the last committed root-session assistant text inside that interval. finish_reason is the kind of the last root turn/end, such as completed, max-tokens, or error, or None when no turn ended.

Neither field is a prompt-causality guarantee. Steering, injected context, and queued work may contribute before idle. Judge automation with the durable events, expected tool results, external state, and finish_reason together. A turn/end without a string reason kind is a runtime protocol violation and raises SdkProtocolError rather than inventing a result.

python
result = session.run('Run the focused tests.')
print(result.session_id, result.finish_reason)
print(result.final_response)
for event in result.events:
    if event.get('type') in {'tool/call', 'tool/result', 'turn/end'}:
        inspect_redacted(event)

Consume root events and descendant notifications correctly

RunResult.events includes durable events for the root session only. RunResult.notifications and on_notification include the root plus known descendant sessions in wire order, including nested subagent lifecycle and session events. HarnessClient retains discovered ancestry for the lifetime of its runtime, so tree filtering remains stable across calls.

python
def observe(notification):
    method = notification.method
    payload = notification.payload
    if method == 'session.event':
        record_redacted(payload.get('sessionId'), payload.get('event'))

result = session.run('Delegate independent checks.', on_notification=observe)
assert all(event_session_is_root(e, result.session_id) for e in result.events)

Descendant messages cannot replace root final_response because that field folds only root events. Callbacks run as notifications are collected; keep them fast, non-throwing, and secret-aware. The low-level session_prompt() returns a queued MessageId immediately and does not wait for activity. Callers using it must define their own receipt, idle, timeout, and subscription boundaries.

Pair client options with the chosen composition

The SDK does not invent provider routes or tools. provider must name an adapter registered by the selected Cordis tree, and model must be resolvable by that adapter. The bundled default registers deepseek-official. A custom tree can mount llm-pi-ai for catalog or company routes, but it must also retain the SDK JSON-RPC server, session services, Agent Loop, persistence, and every provider required by its tools.

Treat the Cordis file as versioned application code. Resolve credential references through the runtime environment or credential provider, keep secrets out of YAML, and validate the composition in the same container that runs Python. A missing provider is a composition defect; changing the Python string until something responds hides rather than fixes it.

yaml
# Conceptual custom composition responsibilities
- sdk JSON-RPC server over stdio
- agent registry and Agent Loop
- chosen LLM adapter and provider route
- session persistence and checkpoint policy
- tool registry plus selected tool providers
- permission and sandbox enforcement
- no console plugin that corrupts protocol stdout

Protocol stdout belongs exclusively to JSON-RPC. Do not add a console logger or application that prints arbitrary text to the runtime's stdout. Put deployment diagnostics on stderr or a structured telemetry path, and test startup with credentials absent so failure remains understandable without exposing them.

Bound requests and classify errors

request_timeout_seconds bounds JSON-RPC requests and interval waits at the client boundary. Choose a value long enough for tools and model turns, plus an outer job budget for the whole workflow. A client timeout does not prove the runtime or an external side effect stopped; inspect durable state before retrying and close the runtime when abandoning the job.

python
from deepseek_harness import DeepSeekHarness

with DeepSeekHarness(
    cwd='/work/repo',
    session_root='/work/sessions',
    request_timeout_seconds=600,
) as harness:
    try:
        result = harness.run('Run integration tests.', session_id='ci-042')
    except Exception as exc:
        classify_sdk_error(exc)
        raise

Distinguish JSON-RPC response errors with preserved code/data, request timeout, protocol violations, and transport closure caused by runtime exit. Provider and tool outcomes generally remain in session events and finish_reason. Log exception class, stable code, runtime exit code, bounded stderr tail, session id, and SDK version after redaction; never log credentials or entire environment snapshots.

Close the runtime on every path

The context manager is the preferred lifecycle because __exit__ closes the reused subprocess even when user code raises. Long-lived services may call close() explicitly in finally. Closing ends the runtime and its sessions, subscriptions, persistent shells, and owned resources; a closed harness is not a pool entry to revive. Create a new instance for later work.

python
harness = DeepSeekHarness(cwd='/work/repo', session_root='/work/sessions')
try:
    result = harness.run('Inspect the project.', session_id='inspect-001')
finally:
    harness.close()

Do not leave cleanup to interpreter shutdown, especially in test runners, notebooks, workers, or services that create repeated clients. On interruption, stop scheduling calls, retain the session id, close once, and wait for process termination according to your supervisor budget. Parallel use should follow the SDK's documented client contract rather than sharing one Session from arbitrary threads.

Use session persistence for recovery evidence

session_root stores durable JSONL and state for every session identity. Before a rollout, run one task, close the harness, construct a new harness against the same root, and intentionally continue the recorded session id. Confirm that history is reconstructed and new sequence values remain contiguous. This tests persistence and runtime restart rather than only in-process Session reuse.

python
with DeepSeekHarness(cwd=workspace, session_root=sessions) as first:
    before = first.run('Record the current branch.', session_id='resume-001')

with DeepSeekHarness(cwd=workspace, session_root=sessions) as second:
    after = second.run('Recall the branch and verify it again.', session_id='resume-001')

assert before.session_id == after.session_id

Back up representative roots before upgrading preview versions, and test migration or rollback on copies. Never repair a failed run by deleting JSONL rows or reusing a session id for an unrelated workspace. Session content can retain secrets emitted by tools or models even when application logs are clean, so encrypt, restrict, redact, and expire it according to the workload.

Treat the minimal example as danger-full-access

The checked-in minimal composition mounts persistent Bash, str_replace_editor, local PTY, bare fs-local, and danger-full-access. Absolute editor paths and shell commands can modify any path visible to the runtime process. cwd is contextual workspace selection, not an enforcement boundary in this composition. The example intentionally omits approval UI and many production controls.

text
Safe evaluation envelope
container or disposable VM
ephemeral checkout with no host mounts
job-scoped API key
restricted network egress
non-root runtime user
private, bounded session storage
no production cloud credentials
reviewed Cordis composition

For production, compose sandboxed filesystem and subprocess providers, narrow model-facing tools, apply deterministic unattended permission policy, isolate network and credentials, and validate the resolved config. Prompt instructions such as ‘do not edit’ are not security controls. Rotate credentials and discard the environment after adversarial tests.

Verify the complete client contract

Test platform installation, lazy startup, first request, same-session continuation, fresh-session isolation, descendant notifications, max-token and error finish reasons, timeout, transport death, malformed protocol, callback failure handling, explicit close, context-manager close, and persistence reload. Confirm the runtime and SDK distributions have the same version.

text
Release checklist
[ ] supported OS, architecture, and Python version
[ ] SDK/runtime wheel versions match
[ ] cwd and session_root are absolute and protected
[ ] session_id reuse is intentional
[ ] RunResult interpreted as an idle-bounded interval
[ ] root events separated from descendant notifications
[ ] timeout and error classes retained
[ ] close runs on success, exception, and interruption
[ ] Cordis tool, persistence, permission, and sandbox rows reviewed
[ ] minimal danger-full-access runs only in isolation

Pin package versions and the Cordis composition together. Keep representative redacted session fixtures for careful automated upgrade compatibility testing, but do not treat preview event shapes as permanent. The SDK is a transport and lifecycle client for the Harness runtime; production reliability comes from pairing it with a deliberate composition, bounded environment, and evidence-based outcome checks.

Official sources