class CursorSdkDriver:
name = "cursor"
supports_logprobs = False
def __init__(
self,
*,
model: str = "",
workspace: str | os.PathLike[str] | None = None,
mcp_servers: dict[str, Any] | None = None,
api_key: str = "",
**_ignored: Any,
) -> None:
try:
import cursor_sdk # noqa: F401
except ImportError as exc: # pragma: no cover - optional dep
raise RuntimeError(
"CursorSdkDriver requires the optional 'cursor-sdk' package and a "
"running Cursor backend. The portable path (ClaudeDriver / "
"GptDriver) needs neither."
) from exc
from dashboard_common.cursor_bridge import (
harden_cursor_sdk_auth_tokens,
resolve_mcp_servers,
)
harden_cursor_sdk_auth_tokens()
self.default_model = model or os.environ.get("ANGELO_CURSOR_MODEL", _DEFAULT_MODEL)
self._workspace = str(Path(workspace) if workspace is not None else Path.cwd())
self._api_key = api_key
servers = (
mcp_servers
if mcp_servers is not None
else {"angelo-zettelkasten": {"command": "angelo-zettelkasten"}}
)
self._mcp_servers = resolve_mcp_servers(servers)
# Single-flight: serialize every run so concurrent executor waves don't
# collide on the one bridge (the bridge cannot run two agents at once).
self._lock = threading.Lock()
self._loop_thread: _LoopThread | None = None
self._client: Any = None
self._model_selection: dict[str, Any] | None = None
atexit.register(self.close)
# -- public protocol ----------------------------------------------------
def run_agent(
self,
*,
role: str,
persona: str,
prompt: str,
tools: list[Tool],
model: str = "",
ctx: dict[str, Any] | None = None,
) -> AgentResult:
# ``tools`` is intentionally ignored: the agent reaches the zettelkasten
# store via the angelo-zettelkasten MCP server, not these callables.
with self._lock:
try:
return self._loop().run(self._run_agent_async(role, persona, prompt, model))
except Exception as exc: # startup failure / bridge crash
detail = getattr(exc, "message", None) or str(exc)
return AgentResult(
text=f"RESULT: FAIL\nNOTES: cursor driver error: {detail}",
ok=False,
)
def close(self) -> None:
"""Dispose the bridge and stop the background loop. Idempotent."""
loop = self._loop_thread
client = self._client
self._client = None
self._model_selection = None
if loop is None:
return
try:
if client is not None:
try:
loop.run(self._shutdown_client(client))
except Exception:
pass
finally:
loop.close()
self._loop_thread = None
# -- async internals ----------------------------------------------------
def _loop(self) -> _LoopThread:
if self._loop_thread is None:
self._loop_thread = _LoopThread()
return self._loop_thread
async def _ensure_client(self) -> Any:
from cursor_sdk import AsyncClient
from dashboard_common.cursor_bridge import read_api_key
# Read the key first so a missing key raises before spinning up a bridge.
api_key = self._api_key or read_api_key(self._workspace)
if self._client is None:
self._client = await AsyncClient.launch_bridge(
workspace=self._workspace,
local={"cwd": self._workspace},
)
return api_key
async def _resolve_model(self, api_key: str, model: str) -> dict[str, Any]:
from dashboard_common.cursor_bridge import resolve_model_selection
chosen = model or self.default_model
# Cache only when the task uses the driver's default model, since a
# per-task override may differ from run to run.
if not model and self._model_selection is not None:
return self._model_selection
selection = await resolve_model_selection(self._client, chosen, api_key)
if not model and selection.get("params"):
self._model_selection = selection
return selection
async def _run_agent_async(
self, role: str, persona: str, prompt: str, model: str
) -> AgentResult:
from cursor_sdk import CursorAgentError
api_key = await self._ensure_client()
model_selection = await self._resolve_model(api_key, model)
system = persona
if role not in _WRITE_ROLES:
system = f"{persona}{_READONLY_GUARD}"
full_prompt = f"{system}\n\n{prompt}" if system else prompt
agent = await self._client.create_agent(
{
"model": model_selection,
"api_key": api_key,
"name": f"stream-{role or 'agent'}",
"local": {"cwd": self._workspace},
"mcp_servers": self._mcp_servers,
}
)
try:
run = await agent.send(full_prompt)
accumulated = ""
tool_calls = 0
seen_tools: set[str] = set()
async for event in run.events():
sdk_message = getattr(event, "sdk_message", None)
if sdk_message is None:
continue
mtype = getattr(sdk_message, "type", "")
if mtype == "assistant":
message = getattr(sdk_message, "message", None)
content = getattr(message, "content", ()) if message else ()
text = "".join((getattr(b, "text", "") or "") for b in content)
accumulated += text
elif mtype == "tool_call":
call_id = getattr(sdk_message, "call_id", "") or str(tool_calls)
if call_id not in seen_tools:
seen_tools.add(call_id)
tool_calls += 1
result = await run.wait()
if getattr(result, "status", "") == "error":
return AgentResult(
text="RESULT: FAIL\nNOTES: cursor run failed mid-flight.",
ok=False,
tool_calls=tool_calls,
)
final = getattr(result, "result", "") or accumulated
return AgentResult(text=final, ok=True, tool_calls=tool_calls, raw=result)
except CursorAgentError as exc:
detail = getattr(exc, "message", None) or str(exc)
return AgentResult(
text=f"RESULT: FAIL\nNOTES: cursor run error: {detail}", ok=False
)
finally:
try:
await agent.close()
except Exception:
pass
@staticmethod
async def _shutdown_client(client: Any) -> None:
try:
await client.shutdown()
except Exception:
pass
try:
await client.aclose()
except Exception:
pass