sddesk 0.5.0

Python SDK for the sdvordesk agent (HTTP + WebSocket, /goal, skills, MCP)

sddesk-sdk

Python SDK for the sdvordesk agent. Mirrors the Cursor SDK Agent / Run model over the sdvordesk HTTP and WebSocket protocols.

Status: v0.5.0 — HTTP chat + WebSocket full-fidelity agent (/goal, tools, skills, MCP) + local runtime via headless server + async mirror. New examples: meet-viewer memory agent, local Context7 MCP. Streaming events include first-class thinking (reasoning) separate from text (assistant response), aligned with the Cursor SDK.

Documentation

Doc	Audience
GitBook (опубликовано)	Live docs на GitBook
GitBook (исходники)	Markdown в репозитории
docs/guide.md	Getting started (RU) — mental model, patterns, traps
docs/api.md	Full API reference
docs/gitbook/publish/gitbook-setup.md	Publish to GitBook (gitbook dev / gitbook publish)
.cursor/skills/sddesk-sdk/	Cursor Agent skill (copy into your project or use from this repo)
examples/	Runnable scripts (local_context7_agent.py, meet_viewer_memory_agent.py, …)

Install

pip install sddesk
# WebSocket (/goal, tools, skills, MCP):
pip install "sddesk[ws]"

PyPI: https://pypi.org/project/sddesk/

Quick start

One-shot chat

from sddesk import Agent

result = Agent.prompt("Fix the bug in auth.py", api_key="sdv_...")
print(result.status, result.content)

Durable agent with streaming

from sddesk import Agent

with Agent.create(api_key="sdv_...") as agent:
    run = agent.send("Refactor auth.py", stream=True)
    for event in run.stream():
        if event.type == "text":
            print(event.text, end="")
        elif event.type == "thinking":
            print(event.text, end="", flush=True)  # reasoning stream (WS only)
    result = run.wait()  # RunResult(status, content, session_id, model)

    # Follow-up keeps conversation context automatically.
    run2 = agent.send("Now write regression tests")
    print(run2.wait().content)

Resume an existing session

from sddesk import Agent

with Agent.resume(session_id="uuid", api_key="sdv_...") as agent:
    agent.send("Update the changelog").wait()

WebSocket + /goal autonomous loop

from sddesk import Agent

with Agent.create(api_key="sdv_...", ws=True) as agent:
    agent.auto_approve_permissions()        # or agent.on_permission(custom_cb)
    run = agent.send("Implement the fizzbuzz feature", stream=True)
    agent.set_goal("npm test exits 0")      # agent loops until the goal is met
    for event in run.stream():
        if event.type == "text":   print(event.text, end="")
        elif event.type == "tool": print(f"\n[tool: {event.name}]")
        elif event.type == "goal": print(f"\n[goal update]")
    result = run.wait()   # blocks until met/failed/budget; result.iterations

Local runtime (headless server)

Run the agent against a local checkout without Postgres — sessions live under <workspace>/.sdvord/sessions/:

from sddesk import Agent, LocalAgentOptions

with Agent.create(local=LocalAgentOptions(cwd="/path/to/project")) as agent:
    run = agent.send("List files in src/", stream=True)
    for event in run.stream():
        if event.type == "text":
            print(event.text, end="")
    print(run.wait().session_id)  # local-...

# Resume a local session (same cwd as when it was created):
with Agent.resume("local-...", local=LocalAgentOptions(cwd="/path/to/project")) as agent:
    agent.send("Continue").wait()

Requires a built sdvordesk server. Set SDVORD_SERVER_ENTRY to dist-server/server/server.js, or run from the sdvordesk-main checkout (SDVORD_REPO_ROOT).

Low-level API: from sddesk import launch_bridge — returns BridgeConnection with api_key, base_url, and the subprocess handle.

Async (for servers / bots)

import asyncio
from sddesk import AsyncAgent

async def main():
    async with AsyncAgent.create(api_key="sdv_...") as agent:
        run = await agent.send("Refactor auth.py", stream=True)
        async for event in run.stream():
            if event.type == "text":
                print(event.text, end="")
        print(await run.wait())

asyncio.run(main())

Error handling (two categories, like Cursor SDK)

import sys
from sddesk import Agent, CursorAgentError

try:
    run = agent.send(prompt)
    result = run.wait()
    if result.status == "error":
        sys.exit(2)  # run executed but failed
except CursorAgentError as e:
    print(f"startup failed: {e} (retryable={e.is_retryable})", file=sys.stderr)
    sys.exit(1)

Auth

The SDK uses a single sdv_* API key for both HTTP and WebSocket. Resolution order:

API key: explicit api_key= → SDESK_API_KEY → CURSOR_API_KEY (alias)

Base URL: explicit base_url= → SDESK_BASE_URL → https://sdvordesk-main.llmlabs.itlabs.io (internal contour)

For a locally running server: export SDESK_BASE_URL=http://localhost:3000

Copy .env.example → .env and set SDESK_API_KEY. The internal URL is not a secret — it is reachable only inside the contour.

Create keys via the sdvordesk UI (POST /api/api-keys) or the REST API. Keys have scopes:

Scope	Allows
chat (default)	HTTP /v1/* — chat completions, files, conversations
ws:run	WebSocket agent runs (sessions, streaming, /goal) — no admin events
ws:admin	Full WebSocket protocol (MCP, scheduler, settings, memory) — grant explicitly

Skills & MCP (WebSocket)

Skills and MCP are server-side (marketplace + per-user MCP store), not local files like Cursor. The SDK exposes them as resources on a WS-connected agent.

Skills per run

HTTP: pass skill_ids= to Agent.prompt / agent.send (maps to sdvordesk.skill_ids).

WebSocket: pass skill_ids= to agent.send or set defaults on Agent.create(ws=True, skill_ids=[...]) (maps to session.start / session.continue skillIds → session metadata.apiSkillIds).

with Agent.create(api_key="sdv_...", ws=True, skill_ids=["using-superpowers"]) as agent:
  run = agent.send("Use the brainstorming skill")
  print(run.wait().content)

Skills marketplace (ws:run list / ws:admin manage)

with Agent.create(api_key="sdv_...", ws=True) as agent:
    catalog = agent.skills.list()           # skills.get → skills.loaded
    fresh = agent.skills.refresh()          # skills.refresh
    # admin scope:
    agent.skills.toggle("skill-id", True)
    agent.skills.install_github("owner/repo/skill-name")
    agent.skills.set_marketplace("https://skills.example.com/index.json")

Inline MCP (ws:admin)

MCP servers are persisted per user on the server. For a Cursor-like bootstrap, pass configs to Agent.create(mcp_servers=[...]) — the SDK calls mcp.add on connect.

mcp = {"name": "ctx7", "type": "mcp", "transport": "http", "url": "https://mcp.example.com/mcp"}

with Agent.create(api_key="sdv_...", ws=True, mcp_servers=[mcp]) as agent:
    servers = agent.mcp.list()
    agent.mcp.test(mcp)
    agent.mcp.import_config('{"mcpServers": {...}}')  # Cursor mcp.json shape

API surface

Method	Description
Agent.prompt(text, ...)	One-shot chat completion (HTTP). Returns RunResult.
Agent.create(...)	Open a durable agent (remote HTTP or WS).
Agent.create(..., local=LocalAgentOptions(cwd=...))	Local headless server + WS bridge (no API key).
launch_bridge(workspace)	Low-level: spawn headless server, return BridgeConnection.
Agent.resume(session_id, ...)	Continue an existing session.
agent.send(text, stream=...)	Send a turn, returns Run.
run.stream() / run.messages()	Iterate StreamEvents (text/tool/permission/goal).
run.wait()	Block until the turn terminates, returns RunResult.
run.text()	Block on wait(), return concatenated assistant text.
run.cancel()	Cancel the run (WS transport).
agent.set_goal(condition)	/goal autonomous loop (WS). run.wait() is goal-aware.
agent.clear_goal()	Deactivate the current goal.
agent.get_goal_status()	Inspect live GoalState.
agent.on_permission(cb) / auto_approve_permissions()	Permission callback (WS).
agent.load_history()	Fetch prior messages for the current session.
agent.skills	Skills marketplace (list, toggle, install_github, set_marketplace).
agent.mcp	MCP server management (list, add, remove, toggle, test, import_config).
agent.list_file_changes()	Files the agent modified in the current session (from session.history).
agent.list_artifacts(path) / read_artifact(path) / download_artifact(path)	Workspace file browser (/api/files*, scope chat).
agent.artifacts	Cursor-style resource wrapping the same /api/files* endpoints.
agent.delete_session(session_id?)	Soft-delete a session (WS only; sends session.delete).
AsyncAgent / AsyncRun	Async mirror (await, async for).

Retries: retryable HTTP errors (429/5xx) are retried with exponential backoff honouring Retry-After. Non-retryable errors (401/403) propagate immediately.

Development

uv sync            # or: pip install -e ".[dev]"
pytest             # unit tests (no live server needed)
ruff check sddesk  # lint
mypy sddesk        # typecheck (strict)

Integration + protocol-drift tests hit a live server and are gated on SDESK_TEST_API_KEY (needs ws:run scope for WS tests):

SDESK_TEST_BASE_URL=https://sdvordesk-main.llmlabs.itlabs.io SDESK_TEST_API_KEY=sdv_... pytest -k "integration or protocol_sync"

tests/test_protocol_sync.py is a drift detector: it runs a WS turn against the live server and fails if the server emits event types the SDK hasn't mirrored — a prompt to update sddesk/protocol/events.py and sddesk/protocol/_version.py.

Cursor Agent skill

This repo ships a Cursor skill at .cursor/skills/sddesk-sdk/. To use it in another project:

cp -r .cursor/skills/sddesk-sdk ~/.cursor/skills/
# or symlink the repo's .cursor/skills into your project

Invoke with @sddesk-sdk or let the agent pick it up from the description when you mention sddesk, SDESK_API_KEY, or Agent.create(ws=True).

License

MIT