cursorconnect 0.1.0

Unofficial Python SDK for Cursor AI agents -- create, stream, orchestrate, and manage cloud and local coding agents from Python.

CursorConnect

The Unofficial Python SDK for Cursor Agents

Orchestrate cloud and local AI coding agents from Python.
One import. Full control. Both runtimes.

Quick Start · Runtimes · Design Philosophy · Examples · API Reference

---

What is CursorConnect?

CursorConnect is the Python interface to Cursor's agent infrastructure. It lets you programmatically create, control, and orchestrate AI coding agents from any Python script, notebook, or backend service, whether those agents run in Cursor's cloud or locally on your own machine.

With CursorConnect you can:

• Spin up agents that read your repos, write code, create PRs, and run commands
• Choose your runtime: cloud agents execute on Cursor's infrastructure; local agents run as a subprocess on your machine with direct filesystem access
• Stream output token-by-token as typed Python objects, including the agent's internal reasoning
• Orchestrate in parallel with the Mesh layer: dispatch multiple specialist agents, let them cross-check each other's work in real time, and collect structured results - New! CursorConnect Only
• Send follow-up instructions to refine work across multiple conversational turns
• Manage your fleet: list, archive, resume, or delete agents programmatically
• Query your account: see available models, connected repositories, and API key metadata

All of this through a clean, Pythonic interface that hides the underlying REST endpoints, SSE framing, Node.js bridge, and HTTP authentication behind three core concepts: Agent, Run, and Mesh.

---

Installation

pip install cursorconnect

Or install from source:

git clone https://github.com/bloomresearch/cursorconnect.git
cd cursorconnect
pip install -e .

Requirements: Python 3.8+ and requests>=2.28 (installed automatically).

For local agents only: Node.js 18+ and the @cursor/sdk npm package must be available on your system. Cloud agents have no additional requirements.

---

Two Runtimes, One Interface

CursorConnect supports two execution modes. The Agent API is the same regardless of which you choose, so you can develop locally and deploy to the cloud (or vice versa) without changing your orchestration code.

Cloud Runtime

Cloud agents run on Cursor's managed infrastructure. You provide a GitHub repo URL, the agent clones it in a sandboxed environment, does its work, and can open a PR when it finishes. This is the zero-setup path: no local dependencies beyond Python and an API key.

from cursorconnect import Agent
from cursorconnect.types import CloudOptions

agent = Agent.create(
    api_key="crsr_...",
    prompt="Add comprehensive type hints to all public functions in src/utils.py",
    cloud=CloudOptions(
        repos=[{"url": "https://github.com/your-org/your-repo"}],
        autoCreatePR=True,
    ),
)

See more examples in the Examples section, or browse the full API Reference.

Local Runtime

Local agents run as a subprocess on your machine via a Node.js bridge to the TypeScript @cursor/sdk. The agent operates directly on your local filesystem, which means it can access private files, work offline, and integrate with local tooling, dev servers, and databases that aren't reachable from the cloud.

from cursorconnect import Agent
from cursorconnect.types import LocalOptions

agent = Agent.create(
    prompt="Refactor the utils module to use pathlib instead of os.path",
    local=LocalOptions(cwd="/path/to/your/project"),
    model="claude-sonnet-4-6",
)

# agent.initial_run returns a RunProtocol (Run or LocalRun)
run = agent.initial_run
if run:
    result = run.wait()
    print(f"Done: {result.status}")

Same Agent class, same Run interface, same stream() and wait() methods. The SDK detects that you passed local instead of cloud and routes through the Node.js bridge automatically. Under the hood, the bridge communicates with the TypeScript SDK over JSON-RPC via stdin/stdout, handles streaming, and automatically restarts if the subprocess crashes.

When to Use Which

	Cloud	Local
Setup	API key only	Node.js + @cursor/sdk
Filesystem	Clones from GitHub	Direct access to local files
Network	Runs on Cursor's infra	Runs on your machine
Best for	CI/CD pipelines, batch operations across repos, PR automation	Interactive development, private codebases, local tool integration
Offline	No	Yes

---

Quick Start

1. Get your API key

Generate a Cursor API key from your Cursor Settings. Set it as an environment variable:

export CURSOR_API_KEY="crsr_your_key_here"

Or pass it directly to any method via the api_key parameter.

2. Fire-and-forget: one prompt, one result

The fastest path from idea to execution. Agent.prompt() creates an agent, sends your message, and hands you back a Run you can wait on:

from cursorconnect import Agent
from cursorconnect.types import CloudOptions

run = Agent.prompt(
    api_key="crsr_...",
    message="Add comprehensive type hints to all public functions in src/utils.py",
    cloud=CloudOptions(
        repos=[{"url": "https://github.com/your-org/your-repo"}],
        autoCreatePR=True,
    ),
)

result = run.wait()
print(f"Done: {result.status}")  # "FINISHED"

# See what the agent actually did
conv = result.conversation
print(conv.text)  # all assistant output as a single string
print(conv)       # full formatted conversation with turns and roles

That's it. The agent clones your repo, makes the changes, opens a PR, and you get a structured result object back. Every RunResult carries a .conversation property that gives you the complete, typed history of the run so you can inspect, log, or post-process what happened.

3. Multi-turn conversation

For iterative workflows, create a persistent agent and send multiple rounds of instructions:

from cursorconnect import Agent
from cursorconnect.types import CloudOptions

agent = Agent.create(
    api_key="crsr_...",
    prompt="Set up a FastAPI project with SQLAlchemy and Alembic",
    cloud=CloudOptions(
        repos=[{"url": "https://github.com/your-org/backend"}],
    ),
)

run1 = agent.send("Add JWT-based authentication with refresh tokens")
result1 = run1.wait()

run2 = agent.send("Now add role-based access control with admin and user roles")
result2 = run2.wait()

# Each result carries the conversation from that run
print(f"Round 1: {len(result1.conversation)} turns")
print(f"Round 2: {len(result2.conversation)} turns")
print(result2.conversation.text)

for artifact in agent.list_artifacts():
    print(f"  {artifact.path}  ({artifact.size_bytes} bytes)")

4. Parallel Agent Mesh

This is what the SDK unlocks that the Cursor UI can't do. In the app, you run one agent at a time. With CursorConnect, you can decompose a feature into subtasks, dispatch parallel agents, and have them cross-check each other's work in real time as each one finishes.

The Mesh is an orchestration layer that manages the full lifecycle: concurrent dispatch, optional peer-to-peer broadcasting, and automatic cleanup. As each agent completes, the Mesh sends a summary of what it built to the agents still running, so they can reconcile conflicts against their own output before they finish. The last agent to complete has seen everyone else's work.

from cursorconnect import Mesh, CommonModels
from cursorconnect.types import CloudOptions, MeshTask

REPO = "https://github.com/your-org/saas-app"

auth_task = MeshTask(
    name="auth-agent",
    prompt=(
        "Implement a complete JWT authentication system: refresh tokens, "
        "middleware, login/register endpoints, and password hashing with bcrypt. "
        "Work only in src/auth/."
    )
)

billing_task = MeshTask(
    name="billing-agent",
    prompt=(
        "Integrate Stripe billing: subscription creation, webhook handlers, "
        "usage metering, and a customer portal redirect endpoint. "
        "Work only in src/billing/."
    )
)

rbac_task = MeshTask(
    name="rbac-agent",
    prompt=(
        "Build a role-based access control system with roles (admin, member, viewer), "
        "permission decorators, and a roles database migration. "
        "Work only in src/rbac/."
    )
)

with Mesh(
    cloud=CloudOptions(repos=[{"url": REPO}], autoCreatePR=True),
    model="claude-sonnet-4-6",
    cross_check=True,
) as mesh:

    mesh + auth_task + billing_task + rbac_task

    print("Dispatching 3 specialist agents in parallel...\n")

    for result in mesh():
        print(f"  {result.name}: {result.run_result.status} "
              f"({len(result.artifacts)} files produced)")

What happens at runtime:

1. All three agents start building simultaneously.
2. The auth-agent finishes first. The Mesh immediately sends a cross-check message to billing and RBAC: "Peer task 'auth-agent' finished... Please review your own work for any conflicts."
3. The billing-agent finishes next. The Mesh notifies the RBAC agent.
4. The rbac-agent finishes last, but by now it has already seen both other agents' summaries. Its final pass reconciles migration ordering, ensures its permission decorators compose correctly with the auth middleware, and avoids route collisions the billing agent introduced.
5. The context manager calls agent.close() on all created agents on exit, ensuring they are archived and resources are released.

The result: a complex multi-system feature built in parallel, where integration bugs are resolved during construction rather than discovered afterward.

---

Design Philosophy

Agents, Runs, and Messages

These three concepts form the backbone of the SDK:

• Agent is a persistent AI coding session. Think of it as a developer you've hired: it has context, memory, and a workspace. You create one, give it instructions, and it gets to work. An agent persists between messages and can handle multiple rounds of conversation.

• Run is a single unit of work within an agent. Every time you send a message, a new run is created (satisfying the RunProtocol). The run streams its progress as events, eventually reaching a terminal state (FINISHED, ERROR, CANCELLED, or EXPIRED). You can stream it, wait for it, or cancel it.

• Messages are the typed events that flow back from a run. Rather than giving you raw JSON, CursorConnect parses every event into a specific Python dataclass. When the agent is reasoning, you get a ThinkingMessage. When it speaks, an AssistantMessage. When it invokes a tool, a ToolUseMessage. Every event exposes a type property that returns its own class, enabling clean event.type is AssistantMessage checks instead of verbose isinstance() calls.

Runtime-Agnostic by Design

The SDK is built around a structural RunProtocol that decouples the agent domain model from the transport layer. The cloud runtime speaks HTTP to Cursor's REST API. The local runtime bridges to the TypeScript SDK via a Node.js subprocess. Both present the same interface, so switching between them doesn't require rewriting your orchestration logic.

Typed All the Way Down

Every API response is deserialized into a Python dataclass with explicit fields, not a nested dictionary you have to spelunk through. Model configurations, repository metadata, conversation turns, run results, and artifacts all have dedicated types with documented fields. This means fewer runtime surprises and full IDE autocompletion from day one.

Minimal Dependencies, Maximum Clarity

The entire SDK depends only on requests for the cloud runtime. No async framework lock-in, no heavy abstractions, no magic. The Agent class is the entry point for agent work. The Cursor namespace handles account-level queries. The Mesh orchestrates parallel execution. The Run gives you full control over individual units of work. Four concepts, and you're productive.

Errors That Tell You What to Do

Every HTTP error is mapped to a specific exception subclass with an is_retryable flag. A RateLimitError tells you to back off. An AuthenticationError tells you to check your key. A ConfigurationError tells you your request was malformed. You never have to parse status codes or guess what went wrong.

Environment-First Authentication

Every method that needs an API key accepts an optional api_key parameter. If you omit it, the SDK automatically reads CURSOR_API_KEY from your environment. This means zero credential management in application code while still allowing explicit overrides for multi-tenant or testing scenarios.

---

Examples

Automated Code Review Bot

Build a bot that reviews PRs and posts feedback:

from cursorconnect import Agent
from cursorconnect.types import CloudOptions

def review_pr(repo_url: str, branch: str):
    """Spin up an agent to review code on a feature branch."""
    run = Agent.prompt(
        message=(
            f"Review the changes on branch '{branch}'. "
            "Focus on security vulnerabilities, performance issues, "
            "and deviation from the project's coding standards. "
            "Provide a summary with severity ratings."
        ),
        cloud=CloudOptions(
            repos=[{"url": repo_url, "startingRef": branch}],
        ),
    )
    result = run.wait(timeout=300)

    if result.status == "FINISHED" and result.conversation:
        print(f"Review complete ({len(result.conversation)} turns):\n")
        # .text gives you all assistant text concatenated across turns
        print(result.conversation.text)
    elif result.status == "ERROR":
        print(f"Review failed: {result.error_message}")

review_pr("https://github.com/your-org/app", "feature/user-auth")

Batch Refactoring Across Multiple Repos

Apply a consistent change across your entire organization:

from cursorconnect import Agent, CommonModels
from cursorconnect.types import CloudOptions

repos = [
    "https://github.com/your-org/service-a",
    "https://github.com/your-org/service-b",
    "https://github.com/your-org/service-c",
]

agents = []
for repo in repos:
    agent = Agent.create(
        prompt=(
            "Upgrade all Python logging to use structured logging with structlog. "
            "Replace print() calls with proper log levels. "
            "Add a structlog configuration module if one doesn't exist."
        ),
        cloud=CloudOptions(
            repos=[{"url": repo}],
            autoCreatePR=True,
        ),
        model="claude-sonnet-4-6",
        name=f"structlog-migration-{repo.split('/')[-1]}",
    )
    agents.append(agent)
    print(f"Started agent for {repo}: {agent.agent_id}")

for agent in agents:
    agent.reload()
    print(f"  {agent.name}: {agent.status}")

Interactive Development Assistant

Build a REPL-style loop for conversational coding:

from cursorconnect import Agent
from cursorconnect.types import CloudOptions, AssistantMessage

agent = Agent.create(
    prompt="You are working on a Python CLI tool. The repo is ready.",
    cloud=CloudOptions(
        repos=[{"url": "https://github.com/your-org/cli-tool"}],
    ),
)
print(f"Agent ready: {agent.agent_id}\n")

while True:
    user_input = input("You: ").strip()
    if user_input.lower() in ("quit", "exit"):
        agent.close()
        print("Session closed.")
        break

    run = agent.send(user_input)
    print("Agent: ", end="")
    for event in run.stream():
        if event.type is AssistantMessage:
            content = event.message.get("content", [])
            for block in content:
                if isinstance(block, dict) and block.get("type") == "text":
                    print(block.get("text", ""), end="")
    print("\n")

Reading Conversation History

After a run completes, retrieve the full conversation as typed Python objects. The Conversation supports print(), .text, iteration, indexing, and len():

from cursorconnect import Agent
from cursorconnect.types import LocalOptions
from cursorconnect.types.conversation import AgentConversationTurn, AssistantContent

agent = Agent.create(
    prompt="Explain the observer pattern in three sentences.",
    local=LocalOptions(cwd="/path/to/project"),
    model="gemini-3-flash",
)

result = agent.initial_run.wait()
conv = result.conversation

# Quick: print the entire conversation
print(conv)

# Just the assistant's text across all turns
print(conv.text)

# Walk the structure for fine-grained access
for turn in conv:
    if isinstance(turn.turn, AgentConversationTurn):
        for step in turn.turn.steps:
            if isinstance(step.message, AssistantContent):
                print(f"Assistant said: {step.message.text[:80]}...")

agent.close()

Downloading Build Artifacts

Retrieve files the agent produced during its run:

from cursorconnect import Agent

agent = Agent.get(api_key="crsr_...", agent_id="bc-xxxxxxxx")

artifacts = agent.list_artifacts()
for artifact in artifacts:
    print(f"{artifact.path} ({artifact.size_bytes} bytes)")
    content = artifact.download_content()
    with open(f"./downloads/{artifact.path}", "wb") as f:
        f.write(content)
    print(f"  Saved to ./downloads/{artifact.path}")

Monitoring with Status Callbacks

React to run state changes as they happen:

from cursorconnect import Agent
from cursorconnect.types import CloudOptions
import time

    run = Agent.prompt(
        message="Run the full test suite and fix any failing tests",
        cloud=CloudOptions(
            repos=[{"url": "https://github.com/your-org/app"}],
        ),
        model="claude-sonnet-4-6",
    )

def on_status(new_status):
    timestamp = time.strftime("%H:%M:%S")
    print(f"[{timestamp}] Run status changed to: {new_status}")

run.on_status_change(on_status)

for event in run.stream():
    pass

print(f"Final status: {run.status}")

Exploring Your Account

Discover what models and repositories are available:

from cursorconnect import Cursor

user = Cursor.me()
print(f"API Key: {user.api_key_name}")
print(f"Email:   {user.user_email}")
print(f"Created: {user.created_at}")

models = Cursor.models.list()
for model in models:
    print(f"  {model.id}: {model.displayName}")
    if model.parameters:
        for param in model.parameters:
            print(f"    param: {param.id} = {param.values}")

repos = Cursor.repositories.list()
for repo in repos:
    visibility = "private" if repo.is_private else "public"
    print(f"  {repo.name} ({visibility}) - {repo.default_branch}")

Agent Fleet Management

List, filter, and clean up your agents:

from cursorconnect import Agent

result = Agent.list(api_key="crsr_...", limit=50, include_archived=False)
print(f"Active agents: {len(result.items)}")

for agent in result.items:
    print(f"  {agent.agent_id} | {agent.name} | {agent.status}")

for agent in result.items:
    if agent.status == "ACTIVE" and "migration" in (agent.name or ""):
        agent.archive()
        print(f"  Archived: {agent.name}")

page = Agent.list(api_key="crsr_...", limit=10)
all_agents = list(page.items)

while page.nextCursor:
    page = Agent.list(api_key="crsr_...", limit=10, cursor=page.nextCursor)
    all_agents.extend(page.items)

print(f"Total agents across all pages: {len(all_agents)}")

---

API Reference

Agent -- The Core Interface

Method	Type	Returns	Description
Agent.create(prompt, *, api_key, cloud, local, model, name)	class	Agent	Create a new agent and enqueue its first run
Agent.prompt(message, *, api_key, cloud, local, model, name)	class	RunProtocol	One-shot: create agent and return the initial run
Agent.get(api_key, agent_id)	class	Agent	Retrieve an existing agent by ID
Agent.resume(api_key, agent_id)	class	Agent	Alias for get (semantic clarity for resumption)
Agent.list(api_key, *, limit, cursor, pr_url, include_archived)	class	ListResult[Agent]	Paginated listing of your agents
agent.send(message, options)	instance	RunProtocol	Send a follow-up message, get a new run
agent.run(message, *, timeout, poll_interval)	instance	RunResult	Send a message and block until completion
agent.reload()	instance	Agent	Refresh metadata from the API (Cloud only)
agent.archive() / agent.unarchive()	instance	None	Toggle archived state (Cloud only)
agent.close()	instance	None	Archive the agent (alias)
agent.delete()	instance	None	Permanently delete the agent (Cloud only)
agent.list_artifacts()	instance	list[Artifact]	List files produced by the agent (Cloud only)
agent.download_artifact(path)	instance	str	Get a presigned download URL (Cloud only)
agent.initial_run	property	RunProtocol | None	The run created alongside the agent

Run -- Execution Control

Method / Property	Returns	Description
run.stream()	Generator[Message]	Yield typed SSE events in real time
run.wait(timeout, poll_interval)	RunResult	Block until the run finishes
run.cancel()	None	Cancel an active run
run.conversation()	Conversation	Full conversation history for this run
run.on_status_change(callback)	None	Register a status-change listener
run.status	str | None	Current status (reactive property)
run.id	str | None	The run's unique identifier

Mesh -- Parallel Orchestration

Method / Property	Returns	Description
Mesh(*, cloud, model, cross_check, cleanup, ...)	Mesh	Create an orchestrator with shared config
mesh.add(task) or mesh + task	Mesh	Register a MeshTask for execution
mesh.run(timeout) or mesh()	list[MeshResult]	Dispatch all tasks and collect results
mesh.results	list[MeshResult]	Access results after execution

Cursor -- Account Operations

Method	Returns	Description
Cursor.me(api_key)	User	Your API key name, email, and creation date
Cursor.models.list(api_key)	list[ModelListItem]	Available models with parameters and variants
Cursor.repositories.list(api_key)	list[Repository]	Connected GitHub repositories

Local Runtime

Local agents use the same Agent class. Pass local=LocalOptions(...) instead of cloud=CloudOptions(...) and the SDK handles everything internally through the Node.js bridge to the TypeScript SDK. No additional imports or setup required.

Artifact -- File Access

Method / Property	Returns	Description
artifact.path	str	Relative file path
artifact.size_bytes	int	File size
artifact.get_download_url()	str	Presigned S3 URL (valid ~15 min)
artifact.download_content()	bytes	Raw file content

---

Streaming Event Types

Every event from run.stream() is a typed dataclass with a type property that returns its own class. This lets you identify events with clean is checks:

if event.type is AssistantMessage:
    ...

Type	What It Represents
AssistantMessage	The agent's written response (text, code, explanations)
ThinkingMessage	The agent's internal reasoning before acting
ToolUseMessage	A tool invocation (file edit, shell command, etc.) with status and result
StatusMessage	A lifecycle transition (RUNNING, FINISHED, ERROR, etc.)
UserMessageEvent	Echo of the user's original message
SystemMessage	System-level metadata (model info, available tools)
TaskMessage	Sub-task progress within the current run
RequestMessage	A request for additional input or confirmation

All event types share agent_id and run_id fields, so you can always identify which agent and run produced them.

Conversation & Structural Types

These are not streaming events but typed containers you work with after a run completes:

Type	What It Represents
Conversation	Typed container for the full history of a run, returned by run.conversation()
ConversationTurn	A single turn wrapping either an AgentConversationTurn or ShellConversationTurn
ConversationStep	One step within an agent turn (assistant message, thinking, or tool call)
AssistantContent	The assistant's text output within a conversation step
ThinkingContent	The model's internal reasoning within a conversation step
ToolCallContent	A tool invocation recorded as a conversation step
RunProtocol	Structural protocol satisfied by both cloud Run and local LocalRun
ModelParameters	Ergonomic builder for model params (e.g. ModelParameters(thinking="high"))

---

Configuration Types

from cursorconnect import CommonModels
from cursorconnect.types import (
    CloudOptions,
    LocalOptions,
    ModelSelection,
    ModelParameters,
    ModelParameterValue,
)

# Cloud: agent runs on Cursor's infrastructure with access to a GitHub repo
cloud = CloudOptions(
    repos=[{"url": "https://github.com/org/repo", "startingRef": "main"}],
    autoCreatePR=True,
    workOnCurrentBranch=False,
)

# Local: agent runs on your machine with direct filesystem access
local = LocalOptions(
    cwd="/path/to/your/project",
    settingSources=["project", "user"],
)

# Model selection: pass a string shorthand or a full ModelSelection
agent1 = Agent.create(
    prompt="Optimize the database queries",
    cloud=cloud,
    model="claude-sonnet-4-6",
)

# Use ModelSelection with thinking shorthand
model = ModelSelection("claude-sonnet-4-6", thinking="high")

# Or build parameters incrementally with ModelParameters
params = ModelParameters()
params.thinking = "high"
agent2 = Agent.create(
    prompt="Optimize the database queries",
    cloud=cloud,
    model=ModelSelection("claude-sonnet-4-6", params=params),
    name="query-optimizer",
)

---

Error Handling

CursorConnect maps every API failure to a specific, actionable exception:

from cursorconnect import Agent
from cursorconnect.exceptions import (
    AuthenticationError,
    RateLimitError,
    ConfigurationError,
    NetworkError,
    CursorAgentError,
)

try:
    agent = Agent.create(api_key="crsr_...", prompt="Hello")
except AuthenticationError:
    print("Bad API key. Check your credentials.")
except RateLimitError as e:
    print(f"Slow down! Retryable: {e.is_retryable}")
except ConfigurationError as e:
    print(f"Bad request: {e.message}")
except NetworkError:
    print("Network issue. Check your connection.")
except CursorAgentError as e:
    print(f"Unexpected error: {e.message} (retryable: {e.is_retryable})")

Exception hierarchy:

CursorAgentError             Base class for all SDK errors
  +-- AuthenticationError     HTTP 401 / 403 - invalid or expired credentials
  +-- RateLimitError          HTTP 429 - too many requests (is_retryable=True)
  +-- ConfigurationError      HTTP 400 - malformed request or missing params
  |     +-- IntegrationNotConnectedError   A required integration is not linked
  +-- NetworkError            Connection failures, timeouts (is_retryable=True)
  +-- UnsupportedRunOperationError   Invalid operation on a terminal run
  +-- UnknownAgentError       HTTP 5xx or unexpected errors (is_retryable=True)

Every exception carries message, is_retryable, code, and cause attributes so you can build robust retry logic without guessing.

---

Testing

The test suite uses pytest and includes both fast unit tests (mocked HTTP) and slow integration tests (live API):

# Run fast unit tests only
pytest -m "not slow"

# Run everything (requires CURSOR_API_KEY in .env or environment)
pytest

To set up your test environment:

pip install -e ".[test]"
echo "CURSOR_API_KEY=crsr_your_key_here" > .env

---

License

MIT License. See LICENSE for details.

---

Built for developers who'd rather write Python than curl commands.