《Cursor文档》-Cursor Python SDK

export CURSOR_API_KEY="your-key"

pip install cursor-sdk

import os

from cursor_sdk import Agent, LocalAgentOptions

with Agent.create(
    model="composer-2.5",
    api_key="crsr_key",
    local=LocalAgentOptions(cwd=os.getcwd()),
) as agent:
    print(agent.send("Summarize what this repository does").text())

from cursor_sdk import Agent, CloudAgentOptions, CloudRepository

with Agent.create(
    model="composer-2.5",
    api_key="crsr_key",
    cloud=CloudAgentOptions(
        repos=[CloudRepository(url="https://github.com/your-org/your-repo", starting_ref="main")],
        auto_create_pr=True,
    ),
) as agent:
    print(agent.send("Add structured logging to the auth middleware").text())

import asyncio
import os

from cursor_sdk import AsyncClient, LocalAgentOptions

async def main():
    async with await AsyncClient.launch_bridge(workspace=os.getcwd()) as client:
        async with await client.agents.create(
            model="composer-2.5",
            api_key="crsr_key",
            local=LocalAgentOptions(cwd=os.getcwd()),
        ) as agent:
            run = await agent.send("Summarize what this repository does")
            print(await run.text())

asyncio.run(main())

from cursor_sdk import Agent, CloudAgentOptions, CloudRepository, LocalAgentOptions

agent = Agent.create(
    model="composer-2.5",
    local=LocalAgentOptions(cwd="."),
)

cloud_agent = Agent.create(
    model="composer-2.5",
    cloud=CloudAgentOptions(
        repos=[CloudRepository(url="https://github.com/your-org/your-repo", starting_ref="main")],
        auto_create_pr=True,
    ),
)

from cursor_sdk import Agent, CloudAgentOptions

with Agent.create(cloud=CloudAgentOptions(repos=[])) as agent:
    run = agent.send("Research the top 3 Python testing frameworks and summarize.")
    print(run.wait().result)

import os

agent = Agent.create(
    model="composer-2.5",
    cloud=CloudAgentOptions(
        repos=[CloudRepository(url="https://github.com/your-org/your-repo")],
        env_vars={
            "STAGING_API_TOKEN": os.environ["STAGING_API_TOKEN"],
        },
    ),
)

from cursor_sdk import Agent, CloudAgentOptions, CloudRepository

with Agent.create(
    model="composer-2.5",
    cloud=CloudAgentOptions(
        repos=[CloudRepository(url="https://github.com/your-org/your-repo")],
        metadata={
            "end_user_id": "user-123",
            "ticket_id": "ENG-456",
        },
    ),
) as agent:
    print(agent.agent_id)

from cursor_sdk import Agent, LocalAgentOptions, ModelParameterValue, ModelSelection

agent = Agent.create(
    model=ModelSelection(
        id="composer-2.5",
        params=[ModelParameterValue(id="fast", value="true")],
    ),
    local=LocalAgentOptions(cwd="."),
)

import os

from cursor_sdk import Agent, LocalAgentOptions, ModelParameterValue, ModelSelection

with Agent.create(
    model=ModelSelection(
        id="auto-smart",
        params=[ModelParameterValue(id="optimize_for", value="balanced")],
    ),
    local=LocalAgentOptions(cwd=os.getcwd()),
) as agent:
    run = agent.send("Find and fix the failing authentication test")
    result = run.wait()

    print(result.status)

from cursor_sdk import Cursor, ModelParameterValue, ModelSelection

models = Cursor.models.list()
router = next((model for model in models if model.id == "auto-smart"), None)
optimize_for = next(
    (
        parameter
        for parameter in (router.parameters if router else [])
        if parameter.id == "optimize_for"
    ),
    None,
)

if router is None or optimize_for is None:
    raise RuntimeError(
        "Cursor Router is not available for this API key. "
        "Verify that Router is enabled for the key's team."
    )

requested_mode = "balanced"
allowed_values = {entry.value for entry in optimize_for.values}

if requested_mode not in allowed_values:
    raise RuntimeError(
        f'Router mode "{requested_mode}" is not enabled for this team.'
    )

model = ModelSelection(
    id=router.id,
    params=[ModelParameterValue(id=optimize_for.id, value=requested_mode)],
)

from cursor_sdk import ModelParameterValue, ModelSelection, SendOptions

run = agent.send(
    "Handle this complex migration",
    SendOptions(
        model=ModelSelection(
            id="auto-smart",
            params=[ModelParameterValue(id="optimize_for", value="intelligence")],
        ),
    ),
)

from cursor_sdk import Agent

with Agent.create(
    {
        "api_key": "crsr_key",
        "model": {"id": "composer-2.5"},
        "local": {"cwd": "."},
    }
) as agent:
    ...

class Agent:
    agent_id: str
    model: ModelSelection | None
    client: CursorClient

    def send(
        self,
        message: str | Mapping[str, Any] | UserMessage,
        options: SendOptions | Mapping[str, Any] | None = None,
        *,
        idempotency_key: str | None = None,
    ) -> Run: ...

    def reload(self) -> None: ...
    def close(self) -> None: ...

    def list_messages(
        self, options: Mapping[str, Any] | None = None
    ) -> list[AgentMessage]: ...
    def list_artifacts(self) -> list[SDKArtifact]: ...
    def download_artifact(self, path: str) -> bytes: ...
    def get_usage(self, *, run_id: str | None = None) -> AgentUsage: ...

    def archive(self, options: Mapping[str, Any] | None = None) -> None: ...
    def unarchive(self, options: Mapping[str, Any] | None = None) -> None: ...
    def delete(self, options: Mapping[str, Any] | None = None) -> None: ...

with Agent.create(model="composer-2.5", local=LocalAgentOptions(cwd=".")) as agent:
    print(agent.send("Explain this repository").text())

from cursor_sdk import close_default_client

close_default_client()

Agent.prompt(
    message: str | Mapping[str, Any] | UserMessage,
    options: AgentOptions | Mapping[str, Any] | None = None,
    *,
    client: CursorClient | None = None,
) -> RunResult

from cursor_sdk import Agent, AgentOptions, LocalAgentOptions

result = Agent.prompt(
    "What does the auth middleware do?",
    AgentOptions(model="composer-2.5", local=LocalAgentOptions(cwd=".")),
)
print(result.result)

from cursor_sdk import AgentOptions, AsyncAgent, LocalAgentOptions

result = await AsyncAgent.prompt(
    "What does the auth middleware do?",
    AgentOptions(model="composer-2.5", local=LocalAgentOptions(cwd=".")),
    client=client,
)

from cursor_sdk import CursorClient, LocalAgentOptions

with CursorClient.launch_bridge(workspace=".") as client:
    with client.agents.create(
        model="composer-2.5",
        api_key="crsr_key",
        local=LocalAgentOptions(cwd="."),
    ) as agent:
        print(agent.send("Summarize what this repository does").text())

from cursor_sdk import CursorClient, DefaultHttpxClient

with CursorClient.launch_bridge(
    workspace=".",
    http_client=DefaultHttpxClient(proxy="http://proxy.example.com"),
) as client:
    ...

from cursor_sdk import AsyncClient, DefaultAsyncHttpxClient

async with await AsyncClient.launch_bridge(
    workspace=".",
    http_client=DefaultAsyncHttpxClient(proxy="http://proxy.example.com"),
) as client:
    ...

short = client.with_options(timeout=5.0, max_retries=2)
agent = short.agents.create(model="composer-2.5", local=LocalAgentOptions(cwd="."))

short_async = async_client.with_options(timeout=5.0, max_retries=2)
agent = await short_async.agents.create(model="composer-2.5", local=LocalAgentOptions(cwd="."))

print(agent.send("Find the bug in src/auth.py").text())

# 同一个智能体，完整的对话上下文会保留下来。
print(agent.send("Fix it and add a regression test").text())

run = await agent.send("Find the bug in src/auth.py")
print(await run.text())

run = await agent.send("Fix it and add a regression test")
print(await run.text())

run = agent.send(
    {
        "text": "What's in this screenshot?",
        "images": [{"data": base64_png, "mime_type": "image/png"}],
    }
)

from cursor_sdk import SDKImage, UserMessage

run = agent.send(
    UserMessage(
        text="What's in this screenshot?",
        images=[SDKImage.from_file("screenshot.png")],
    )
)

class Run:
    id: str
    agent_id: str
    status: str  # "running" | "finished" | "error" | "cancelled" | "expired"
    result: str
    model: ModelSelection | None
    duration_ms: int
    git: RunGitInfo | None
    created_at: str | None
    usage: TokenUsage | None  # 累计值；实时句柄的属性

    def stream(self) -> Iterator[SDKMessage]: ...
    def messages(self) -> Iterator[SDKMessage]: ...
    def events(self) -> Iterator[RunStreamEvent]: ...
    def iter_text(self) -> Iterator[str]: ...
    def text(self) -> str: ...
    def wait(self) -> RunResult: ...
    def cancel(self) -> None: ...
    def conversation(self) -> list[ConversationTurn]: ...
    def conversation_json(self) -> str: ...
    def observe(self, *, after_offset: str | None = None) -> Iterator[RunStreamEvent]: ...

    def supports(self, operation: str) -> bool: ...
    def unsupported_reason(self, operation: str) -> str | None: ...
    def on_did_change_status(
        self, listener: Callable[[str], None]
    ) -> Callable[[], None]: ...

run = agent.send("Find the bug in src/auth.py")

for message in run.messages():
    if message.type == "assistant":
        for block in message.message.content:
            if block.type == "text":
                print(block.text, end="")
    elif message.type == "thinking":
        print(message.text, end="")
    elif message.type == "tool_call":
        print(f"[tool] {message.name}: {message.status}")
    elif message.type == "status":
        print(f"[status] {message.status}")
    elif message.type == "usage":
        print(f"[usage] turn total={message.usage.total_tokens}")

result = run.wait()

print(result.status)       # "finished" | "error" | "cancelled" | "expired"
print(result.result)       # 最终助手文本（如有）
print(result.model)        # 此次运行所用的已解析 ModelSelection
print(result.duration_ms)
print(result.usage)        # 累计 TokenUsage；不可用时为 None
print(result.git)          # 云端 RunGitInfo

result = await run.wait()

@dataclass(frozen=True)
class TokenUsage:
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int
    cache_write_tokens: int
    total_tokens: int
    reasoning_tokens: int | None = None

result = run.wait()

if result.usage is not None:
    print(f"total: {result.usage.total_tokens}")
    print(f"in: {result.usage.input_tokens}, out: {result.usage.output_tokens}")
    print(
        f"cache read/write: {result.usage.cache_read_tokens}/{result.usage.cache_write_tokens}"
    )
else:
    print("no usage reported for this run")

for message in run.messages():
    if message.type == "usage":
        print(f"turn used {message.usage.total_tokens} tokens")

# 或在等待完成后 / 无需自行读取消息：
result = run.wait()
print(run.usage, result.usage)

for chunk in run.iter_text():
    print(chunk, end="")

final_text = run.text()

async for chunk in run.iter_text():
    print(chunk, end="")

final_text = await run.text()

run.cancel()

await run.cancel()

if run.status == "running":
    run.cancel()

print(run.id)
print(run.status)  # "running" | "finished" | "error" | "cancelled" | "expired"

stop = run.on_did_change_status(lambda status: print(f"status changed to {status}"))
stop()  # 移除监听器

turns = run.conversation()

from cursor_sdk import ModelParameterValue, ModelSelection, SendOptions

run = agent.send(
    "Plan the refactor",
    SendOptions(
        model=ModelSelection(
            id="composer-2.5",
            params=[ModelParameterValue(id="fast", value="true")],
        ),
    ),
)

from cursor_sdk import CloudSendOptions, SendOptions

run = agent.send(
    "Deploy the preview environment",
    SendOptions(
        cloud=CloudSendOptions(env_vars={"DEPLOY_TOKEN": mint_short_lived_token()}),
    ),
)

from cursor_sdk import Agent, AgentOptions, CloudAgentOptions, CloudRepository, SendOptions

with Agent.create(
    AgentOptions(
        model="composer-2.5",
        mode="plan",
        cloud=CloudAgentOptions(
            repos=[CloudRepository(url="https://github.com/your-org/your-repo")],
        ),
    )
) as agent:
    agent.send("Design the auth refactor").wait()
    agent.send(
        "Looks good, start building",
        SendOptions(mode="agent"),
    ).wait()

from cursor_sdk import SendOptions

def on_delta(update):
    if update.type in ("text-delta", "thinking-delta"):
        print(update.text, end="")

run = agent.send(
    "Refactor the utils module",
    SendOptions(on_delta=on_delta, on_step=lambda step: print(f"[step] {step.type}")),
)
run.wait()

from cursor_sdk.events import TextDeltaUpdate, ToolCallStartedUpdate

if isinstance(update, TextDeltaUpdate):
    print(update.text)

SDKMessage = (
    SDKSystemMessage
    | SDKUserMessageEvent
    | SDKAssistantMessage
    | SDKThinkingMessage
    | SDKToolUseMessage
    | SDKStatusMessage
    | SDKTaskMessage
    | SDKRequestMessage
    | SDKUsageMessage
    | Mapping[str, Any]
)

@dataclass(frozen=True)
class SDKUsageMessage:
    type: Literal["usage"]
    agent_id: str
    run_id: str
    usage: TokenUsage

for event in run.events():
    print(event.kind, event.offset)

InteractionUpdate = (
    TextDeltaUpdate
    | ThinkingDeltaUpdate
    | ThinkingCompletedUpdate
    | ToolCallStartedUpdate
    | ToolCallCompletedUpdate
    | PartialToolCallUpdate
    | TokenDeltaUpdate
    | StepStartedUpdate
    | StepCompletedUpdate
    | TurnEndedUpdate
    | UserMessageAppendedUpdate
    | SummaryUpdate
    | SummaryStartedUpdate
    | SummaryCompletedUpdate
    | ShellOutputDeltaUpdate
    | UnknownInteractionUpdate
    | Mapping[str, Any]
)

@dataclass(frozen=True)
class ConversationTurn:
    type: str  # "agentConversationTurn" | "shellConversationTurn"
    turn: AgentConversationTurn | ShellConversationTurn | Mapping[str, Any]

@dataclass(frozen=True)
class AgentConversationTurn:
    user_message: Mapping[str, Any] | None = None
    steps: Sequence[ConversationStep] = ()

@dataclass(frozen=True)
class ShellConversationTurn:
    shell_command: ShellCommand | None = None
    shell_output: ShellOutput | None = None

ConversationStep = (
    AssistantConversationStep
    | ToolCallConversationStep
    | ThinkingConversationStep
    | Mapping[str, Any]
)

for turn in run.conversation():
    if turn.type == "agentConversationTurn":
        for step in turn.turn.steps:
            print(step.type)
    elif turn.type == "shellConversationTurn":
        print(turn.turn.shell_command, turn.turn.shell_output)

Agent.resume(
    agent_id: str,
    options: AgentOptions | Mapping[str, Any] | None = None,
    *,
    client: CursorClient | None = None,
) -> Agent

agent = Agent.resume("bc-abc123")
run = agent.send("Also update the changelog")
run.wait()

agent = await client.agents.resume("bc-abc123")
run = await agent.send("Also update the changelog")
await run.wait()

from cursor_sdk import CursorClient

with CursorClient.launch_bridge(workspace="/path/to/repo") as client:
    agents = client.agents.list(runtime="local", cwd="/path/to/repo")
    info = client.agents.get(agents.items[0].agent_id, cwd="/path/to/repo")

from cursor_sdk import CursorClient

with CursorClient.launch_bridge(workspace=".") as client:
    agents = client.agents.list(runtime="local", cwd=".")

    for agent_info in agents.auto_paging_iter():
        print(agent_info.agent_id)

    info = client.agents.get(agents.items[0].agent_id)
    runs = client.agents.list_runs(info.agent_id)
    run = client.agents.get_run(runs.items[0].id)

agents = await client.agents.list(runtime="local", cwd=".")

async for agent_info in agents.auto_paging_iter():
    print(agent_info.agent_id)

info = await client.agents.get(agents.items[0].agent_id)
runs = await client.agents.list_runs(info.agent_id)
run = await client.agents.get_run(runs.items[0].id)

@dataclass(frozen=True)
class AgentMessage:
    type: str
    uuid: str
    agent_id: str
    message: Any = None

@dataclass(frozen=True)
class SDKAgentInfo:
    agent_id: str
    name: str
    summary: str
    last_modified: str | None = None
    status: str | None = None  # "running" | "finished" | "error"
    created_at: str | None = None
    archived: bool = False
    runtime: Literal["local", "cloud"] | None = None
    cwd: str = ""
    env: CloudEnvironment | None = None
    repos: Sequence[str] = ()
    metadata: Mapping[str, str] = {}  # 来自 CloudAgentOptions.metadata；本地智能体为空

# 通过 ID 操作，无需智能体句柄：
Agent.archive(agent_id)
Agent.unarchive(agent_id)
Agent.delete(agent_id)

# 通过显式指定的客户端：
client.agents.archive(agent_id)
client.agents.unarchive(agent_id)
client.agents.delete(agent_id)

# 通过现有智能体句柄：
agent.archive()
agent.unarchive()
agent.delete()

usage = agent.get_usage()

print(f"tokens: {usage.usage.total_tokens}")
if usage.cost is not None:
    print(f"charged: ${usage.cost.charged_cents / 100:.2f}")
for run in usage.runs:
    print(run.run_id, run.usage.total_tokens)

@dataclass(frozen=True)
class AgentUsage:
    usage: TokenUsage              # 所有 `runs` 的汇总
    runs: Sequence[RunUsage] = ()
    cost: UsageCost | None = None  # 所有 `runs` 的汇总

@dataclass(frozen=True)
class RunUsage:
    run_id: str
    usage: TokenUsage
    cost: UsageCost | None = None

@dataclass(frozen=True)
class UsageCost:
    raw_cost_cents: float  # 未享受折扣的模型 token 成本；按请求计费的用量为 0
    charged_cents: float   # 实际收取的金额，已计入折扣和 Cursor Token 费率

from cursor_sdk import Cursor

me = Cursor.me()
models = Cursor.models.list()
repositories = Cursor.repositories.list()

me = client.me()
models = client.models.list()
repositories = client.repositories.list()

from cursor_sdk import AsyncCursor

me = await AsyncCursor.me(client=client)
models = await AsyncCursor.models.list(client=client)
repositories = await AsyncCursor.repositories.list(client=client)

models = Cursor.models.list()
composer = next((model for model in models if model.id == "composer-2.5"), None)

print(composer.parameters if composer else [])
# [
#   ModelParameterDefinition(
#       id="fast",
#       display_name="Fast",
#       values=(
#           ModelParameterDefinitionValue(value="false"),
#           ModelParameterDefinitionValue(value="true", display_name="Fast"),
#       ),
#   ),
# ]

from cursor_sdk import (
    Agent,
    AgentOptions,
    HttpMcpServerConfig,
    LocalAgentOptions,
    McpAuth,
    StdioMcpServerConfig,
)

agent = Agent.create(
    AgentOptions(
        model="composer-2.5",
        local=LocalAgentOptions(cwd="."),
        mcp_servers={
            "docs": HttpMcpServerConfig(
                url="https://example.com/mcp",
                auth=McpAuth(client_id="client-id", scopes=["read", "write"]),
            ),
            "filesystem": StdioMcpServerConfig(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-filesystem", "."],
            ),
        },
    )
)

from cursor_sdk import (
    Agent,
    AgentOptions,
    CloudAgentOptions,
    CloudRepository,
    HttpMcpServerConfig,
    StdioMcpServerConfig,
)

agent = Agent.create(
    AgentOptions(
        model="composer-2.5",
        cloud=CloudAgentOptions(
            repos=[CloudRepository(url="https://github.com/your-org/your-repo")],
        ),
        mcp_servers={
            "linear": HttpMcpServerConfig(
                url="https://mcp.linear.app/mcp",
                headers={"Authorization": "Bearer linear_pat_xxx"},
            ),
            "github": StdioMcpServerConfig(
                command="npx",
                args=["-y", "@modelcontextprotocol/server-github"],
                env={"GITHUB_TOKEN": "ghp_xxx"},
            ),
        },
    )
)

from cursor_sdk import Agent, AgentDefinition, AgentOptions, LocalAgentOptions

agent = Agent.create(
    AgentOptions(
        model="composer-2.5",
        local=LocalAgentOptions(cwd="."),
        agents={
            "code-reviewer": AgentDefinition(
                description="Expert code reviewer for quality and security.",
                prompt="Review code for bugs, security issues, and proven approaches.",
                model="inherit",
            ),
            "test-writer": AgentDefinition(
                description="Writes tests for code changes.",
                prompt="Write comprehensive tests for the given code.",
            ),
        },
    )
)

from cursor_sdk import Agent, AgentOptions, LocalAgentOptions

# 只读智能体：仅提供以下工具。
reader = Agent.create(
    AgentOptions(
        model="composer-2.5",
        tools=["read", "grep", "glob", "ls"],
        local=LocalAgentOptions(cwd="."),
    )
)

# 除 shell 外的所有工具。
no_shell = Agent.create(
    AgentOptions(
        model="composer-2.5",
        disallowed_tools=["shell"],
        local=LocalAgentOptions(cwd="."),
    )
)

from cursor_sdk import Agent, CustomTool, CustomToolContext, LocalAgentOptions

def get_deployment_status(args, context: CustomToolContext):
    service = args["service"]
    return f"Service {service} is healthy."

with Agent.create(
    model="composer-2.5",
    local=LocalAgentOptions(
        cwd=".",
        custom_tools={
            "get_deployment_status": CustomTool(
                description="Look up the current deployment status for a service.",
                input_schema={
                    "type": "object",
                    "properties": {
                        "service": {"type": "string", "description": "Service name"},
                    },
                    "required": ["service"],
                },
                execute=get_deployment_status,
            ),
        },
    ),
) as agent:
    agent.send("Is the checkout service healthy?").wait()

@dataclass(frozen=True)
class SDKArtifact:
    path: str
    size_bytes: int = 0
    updated_at: str = ""

from pathlib import Path

artifacts = agent.list_artifacts()

for artifact in artifacts:
    print(artifact.path, artifact.size_bytes)

# 将单个产物下载到本地磁盘。
content = agent.download_artifact(artifacts[0].path)
Path("review.md").write_bytes(content)

from cursor_sdk import Agent, LocalAgentOptions

with Agent.create(model="composer-2.5", local=LocalAgentOptions(cwd=".")) as agent:
    agent.send("Summarize the repository").wait()

agent.close()

from cursor_sdk import AsyncClient, LocalAgentOptions

async with await AsyncClient.launch_bridge(workspace=".") as client:
    async with await client.agents.create(
        model="composer-2.5",
        local=LocalAgentOptions(cwd="."),
    ) as agent:
        run = await agent.send("Summarize the repository")
        await run.wait()

await agent.close()
await client.aclose()

from cursor_sdk import close_default_client

close_default_client()

@dataclass
class CustomTool:
    execute: Callable[[Mapping[str, Any], CustomToolContext], Any]
    description: str | None = None
    input_schema: Mapping[str, Any] | None = None

class CustomToolContext:
    tool_call_id: str | None = None

@dataclass(frozen=True)
class ModelSelection:
    id: str
    params: Sequence[ModelParameterValue] = ()

@dataclass(frozen=True)
class ModelParameterValue:
    id: str
    value: str

from cursor_sdk.types import McpServerConfig

@dataclass(frozen=True)
class HttpMcpServerConfig:
    url: str
    type: Literal["http", "sse"] | str = "http"
    headers: Mapping[str, str] | None = None
    auth: McpAuth | Mapping[str, Any] | None = None

@dataclass(frozen=True)
class SseMcpServerConfig(HttpMcpServerConfig):
    type: Literal["sse"] = "sse"

@dataclass(frozen=True)
class StdioMcpServerConfig:
    command: str
    args: Sequence[str] | None = None
    env: Mapping[str, str] | None = None
    cwd: str | os.PathLike | None = None  # 仅限本地；云端不支持此字段

@dataclass(frozen=True)
class McpAuth:
    client_id: str
    client_secret: str | None = None
    scopes: Sequence[str] = ()

@dataclass(frozen=True)
class UserMessage:
    text: str
    images: Sequence[SDKImage | Mapping[str, Any]] | None = None

@dataclass(frozen=True)
class SDKImage:
    url: str | None = None
    data: str | None = None
    mime_type: str | None = None
    dimension: SDKImageDimension | Mapping[str, Any] | None = None

    @classmethod
    def from_url(cls, url: str, dimension=None) -> SDKImage: ...

    @classmethod
    def from_data(cls, data: bytes | str, mime_type: str, dimension=None) -> SDKImage: ...

    @classmethod
    def url_image(cls, url: str, dimension=None) -> SDKImage: ...

    @classmethod
    def data_image(cls, data: str, mime_type: str, dimension=None) -> SDKImage: ...

    @classmethod
    def from_file(cls, path, *, mime_type=None, dimension=None) -> SDKImage: ...

from cursor_sdk.types import SettingSource

@dataclass(frozen=True)
class ListResult(Generic[T]):
    items: list[T]
    next_cursor: str = ""

    def to_dict(self) -> dict[str, Any]: ...
    def has_next_page(self) -> bool: ...
    def next_page_info(self) -> dict[str, str]: ...
    def get_next_page(self) -> ListResult[T]: ...
    def auto_paging_iter(self) -> Iterator[T]: ...

class CursorAgentError(Exception):
    message: str
    code: str | None
    status: int | None
    status_code: int | None
    details: list[Mapping[str, Any]]
    is_retryable: bool
    cause: BaseException | None
    proto_error_code: str | None
    request_id: str | None
    headers: Mapping[str, str]
    retry_after: str | None

import time

from cursor_sdk import Agent, AgentOptions, CursorAgentError, LocalAgentOptions, RateLimitError

for attempt in range(3):
    try:
        result = Agent.prompt(
            "Audit the auth middleware for missing input validation",
            AgentOptions(model="composer-2.5", local=LocalAgentOptions(cwd=".")),
        )
        break
    except RateLimitError as err:
        time.sleep(float(err.retry_after) if err.retry_after else 2**attempt)
    except CursorAgentError as err:
        if not err.is_retryable:
            raise
        time.sleep(2**attempt)

class IntegrationNotConnectedError(ConfigurationError):
    provider: str   # 例如："github"、"gitlab"、"azuredevops"
    help_url: str   # 用于重新连接的仪表盘链接

from cursor_sdk import Agent, AgentBusyError

agent = Agent.resume("bc-00000000-0000-0000-0000-000000000001")

try:
    agent.send("Also add tests for the auth middleware.")
except AgentBusyError:
    runs = Agent.list_runs(agent.agent_id, {"runtime": "cloud", "limit": 1})
    active = runs.items[0] if runs.items else None
    if active is not None and active.status == "running":
        active.cancel()
    agent.send("Also add tests for the auth middleware.")

class UnsupportedRunOperationError(ConfigurationError):
    operation: str

CURSOR_SDK_LOG=debug python my_script.py

cursor-sdk-bridge --help