Cortex Code Agent SDK 参考 – Python¶

本主题提供了适用于 Cortex Code Agent SDK for Python 的完整 API 参考，包括所有函数、类和类型。

安装¶

pip install cortex-code-agent-sdk

需要 Python 3.10 或更高版本。依赖项：anyio、mcp、typing_extensions。请以 cortex_code_agent_sdk 名称导入该包。该 SDK 要求另行安装 Cortex Code CLI。如果它不在您的 PATH 中，请设置``CORTEX_CODE_CLI_PATH=/path/to/cortex``，或在 CortexCodeAgentOptions 中传入 cli_path。

函数¶

query()¶

与 Cortex Code 交互的主要函数。返回一个异步迭代器，该迭代器在消息到达时生成消息。

async def query(
    *,
    prompt: str | AsyncIterable[dict],
    options: CortexCodeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]: ...

参数


参数	类型	描述
`prompt`	`str \| AsyncIterable[dict]`	用户提示字符串，或者是用于流式输入的、由消息字典组成的异步迭代器
`options`	`CortexCodeAgentOptions \| None`	配置选项。默认为 `CortexCodeAgentOptions()`。
`transport`	`Transport \| None`	自定义传输。默认为子进程 CLI 运输

一个异步迭代器，生成 Message 对象（AssistantMessage、ResultMessage、UserMessage、SystemMessage、StreamEvent）。

示例

import asyncio
from cortex_code_agent_sdk import query, AssistantMessage, ResultMessage, CortexCodeAgentOptions

async def main():
    async for message in query(
        prompt="Fix the bug in utils.py",
        options=CortexCodeAgentOptions(cwd="/path/to/project"),
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(block.text, end="")
        elif isinstance(message, ResultMessage):
            print(f"\nDone: {message.subtype}")

asyncio.run(main())

CortexCodeSDKClient¶

用于多轮交互式对话。支持异步上下文管理器协议。

from cortex_code_agent_sdk import CortexCodeSDKClient, CortexCodeAgentOptions, AssistantMessage

async with CortexCodeSDKClient(CortexCodeAgentOptions(permission_mode="bypassPermissions", allow_dangerously_skip_permissions=True)) as client:
    # First turn
    await client.query("What files are in this directory?")
    async for msg in client.receive_response():
        if isinstance(msg, AssistantMessage):
            for b in msg.content:
                if hasattr(b, "text"):
                    print(b.text, end="")

    # Second turn (context preserved)
    await client.query("Now refactor the main function")
    async for msg in client.receive_response():
        if isinstance(msg, AssistantMessage):
            for b in msg.content:
                if hasattr(b, "text"):
                    print(b.text, end="")

方法¶


方法	描述
`connect()`	连接到 CLI 进程，并保持会话开启以支持后续多轮交互。
`query(prompt, session_id?)`	向代理发送提示
`receive_messages()`	生成来自代理的所有消息
`receive_response()`	生成消息，直到 ``ResultMessage``（含）
`interrupt()`	发送中断信号
`set_permission_mode(mode)`	为对话中的后续轮次更改权限模式
`set_model(model)`	在对话期间更改模型
`stop_task(task_id)`	停止正在运行的子代理任务
`disconnect()`	断开与 CLI 进程的连接

选项¶

传递给 query() 或 CortexCodeSDKClient 的配置。


选项	类型	默认值	描述
`cwd`	`str \| Path \| None`	`None`	会话的工作目录
`model`	`str \| None`	`None`	要使用的模型。使用 `"auto"` 进行自动选择，或使用特定标识符，如 `"claude-sonnet-4-6"`。请参阅支持的模型。
`connection`	`str \| None`	`None`	来自 Snowflake CLI 连接设置的 Snowflake 连接名称，通常为 `~/.snowflake/connections.toml`，对于现有设置，也支持 `~/.snowflake/config.toml`。如果省略，则 CLI 使用来自 TOML 文件的 `default_connection_name`。请参阅配置连接。
`profile`	`str \| None`	`None`	配置文件名称（加载自 `~/.snowflake/cortex/profiles/`）
`permission_mode`	`PermissionMode \| None`	`None`	`"default"` \| `"autoAcceptPlans"` \| `"plan"` \| `"bypassPermissions"`。注意：`"bypassPermissions"` 需要 `allow_dangerously_skip_permissions=True`。在 `"plan"` 模式下，`AskUserQuestion` 和 `ExitPlanMode` 可以通过 `can_use_tool` 路由；拒绝 `ExitPlanMode` 会使规划保持活动状态，而批准它会退出规划模式，以便后续轮次恢复正常权限。
`allow_dangerously_skip_permissions`	`bool`	`False`	使用 `permission_mode="bypassPermissions"` 时需要安全标志。该标志本身不会绕过权限；仅在通过 permission_mode 明确请求时才允许绕过权限。
`allowed_tools`	`list[str]`	`[]`	无需提示即可自动批准的工具
`disallowed_tools`	`list[str]`	`[]`	始终拒绝的工具
`max_turns`	`int \| None`	`None`	限制每个查询中代理轮次次数
`effort`	`str \| None`	`None`	模型执行强度等级：`"minimal"` \| `"low"` \| `"medium"` \| `"high"` \| `"max"`
`system_prompt`	`str \| SystemPromptPreset \| None`	`None`	自定义系统提示。传入字符串以完全替换，或传入 `{"type": "preset", "append": "extra text"}` 以附加在默认提示之后。
`continue_conversation`	`bool`	`False`	继续最近的会话，而不是开始新会话
`resume`	`str \| None`	`None`	用于恢复先前对话的会话 ID
`fork_session`	`bool`	`False`	恢复时，派生出一个新的会话 ID，而不是在原会话上继续
`add_dirs`	`list[str \| Path]`	`[]`	要添加到代理上下文的其他目录
`env`	`dict[str, str]`	`{}`	传递给 CLI 进程的环境变量
`plugins`	`list[SdkPluginConfig]`	`[]`	插件配置。目前支持本地插件：`{"type": "local", "path": "/path/to/plugin"}`
`abort_event`	`asyncio.Event \| None`	`None`	一个 `asyncio.Event` 对象，一旦设置完成，将向正在运行的代理发送中断信号。会话保持活动状态，以供后续提示使用。这是 TypeScript SDK 中 `AbortController` 的 Python 等效实现。
`mcp_servers`	`dict[str, McpServerConfig]`	`{}`	外部 MCP 服务器配置。传入一个将服务器名称映射到 stdio、HTTP 或 SSE 配置的字典。当前 SDK 传输支持基于字典的 MCP 配置。
`hooks`	`dict \| None`	`None`	Hook 事件处理程序（请参阅 Hook）
`can_use_tool`	`CanUseTool \| None`	`None`	自定义工具权限回调（请参阅工具权限）
`include_partial_messages`	`bool`	`False`	包括词元级流事件 (`StreamEvent`)
`output_format`	`dict \| None`	`None`	结构化输出格式。示例：`{"type": "json_schema", "schema": {...}}`
`no_mcp`	`bool`	`False`	禁用 MCP 服务器
`session_id`	`str \| None`	`None`	要使用的显式会话 ID
`setting_sources`	`list[str] \| None`	`None`	要加载的设置来源：`"user"` \| `"project"` \| `"local"`
`cli_path`	`str \| Path \| None`	`os.environ.get("CORTEX_CODE_CLI_PATH") or "cortex"`	CLI 二进制文件的路径。如果省略，SDK 会首先检查 `CORTEX_CODE_CLI_PATH`，否则回退到 `PATH` 上的 `cortex`。
`extra_args`	`dict[str, str \| None]`	`{}`	作为键值对的额外 CLI 标志。对于布尔标志，请使用 `None`。
`stderr`	`Callable[[str], None] \| None`	`None`	每当 CLI 的标准错误输出一行时，就会调用该回调函数。

消息类型¶

AssistantMessage¶

当代理产生响应时发出。包含一个或多个内容块。

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None

AssistantMessageError 是以下之一："authentication_failed"、"billing_error"、"rate_limit"、"invalid_request"、"server_error"、"unknown"。

ResultMessage¶

当代理完成一次轮转时发出。检查 subtype 和 is_error 来判断成功还是失败。

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None

UserMessage¶

处理用户消息时回显。

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None
    tool_use_result: dict[str, Any] | None = None

SystemMessage¶

系统事件，例如会话初始化和任务更新。

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

SDK 还为与任务相关的系统消息提供专门的子类：


子类	描述
`TaskStartedMessage`	子代理任务启动时发出。字段：`task_id`、`description`、`uuid`、`session_id`、`tool_use_id`、`task_type`。
`TaskProgressMessage`	在任务运行时发出。字段：`task_id`、`description`、`usage` (`TaskUsage`)、`uuid`、`session_id`、`last_tool_name`。
`TaskNotificationMessage`	当任务完成、失败或停止时发出。字段：`task_id`、`status` (`"completed"` \| `"failed"` \| `"stopped"`)、`output_file`、`summary`、`uuid`、`session_id`、`usage`。

这些子类扩展自 SystemMessage，因此现有的 isinstance(msg, SystemMessage) 检查仍然可以进行匹配。

StreamEvent¶

词元级别流式传输期间的部分消息更新。需要 include_partial_messages=True。

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]   # Partial text/thinking stream event from Cortex Code
    parent_tool_use_id: str | None = None

针对部分文本和思维块发出 StreamEvent。完整的工具调用仍然以 AssistantMessage 内容块的形式到达，工具结果仍然以 UserMessage 内容块的形式到达。

内容块¶


类型	字段
`TextBlock`	`.type = "text"`、`.text: str`
`ThinkingBlock`	`.type = "thinking"`、 `.thinking: str`、 `.signature: str`
`ToolUseBlock`	`.type = "tool_use"`、`.id: str`、`.name: str`、`.input: dict`
`ToolResultBlock`	`.type = "tool_result"`、`.tool_use_id: str`、`.content: str \| list \| None`、`.is_error: bool \| None`

Hook¶

Hook 允许您在生命周期事件发生时进行拦截，用于日志记录、验证或自定义行为。

配置 Hook¶

Hook 通过 CortexCodeAgentOptions 中的 hooks 选项进行配置。每个 Hook 事件对应一个 HookMatcher 对象列表。

from cortex_code_agent_sdk import CortexCodeAgentOptions, HookMatcher

async def my_pre_tool_hook(input_data, tool_use_id, context):
    print(f"Tool {input_data['tool_name']} about to run")
    return {"continue_": True}

options = CortexCodeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",          # Only match Bash tool, or None for all
                hooks=[my_pre_tool_hook],
                timeout=30.0,            # Timeout in seconds (default: 60)
            ),
        ],
    },
)

Hook 回调签名¶

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]

输入：强类型 Hook 输入（见下表）
tool_use_id：可选的工具使用标识符
context：包含 signal 字段的 ``HookContext``（保留用于将来的中止信号支持）

Hook 事件¶


事件	输入类型	描述
`PreToolUse`	`PreToolUseHookInput`	在执行工具之前。字段：`tool_name`、`tool_input`、`tool_use_id`。
`PostToolUse`	`PostToolUseHookInput`	工具完成后。字段：`tool_name`、`tool_input`、`tool_response`、`tool_use_id`。
`UserPromptSubmit`	`UserPromptSubmitHookInput`	用户提交提示时。字段：`prompt`。
`Stop`	`StopHookInput`	当代理停止时。字段：`stop_hook_active`。
`SubagentStop`	`SubagentStopHookInput`	子代理完成时。字段：`stop_hook_active`。
`Notification`	`NotificationHookInput`	发生通知事件时。字段：`message`、`notification_type`。
`PermissionRequest`	`PermissionRequestHookInput`	工具请求权限时。字段：`tool_name`、`tool_input`。
`PreCompact`	`PreCompactHookInput`	上下文压缩之前。字段：`trigger` (`"manual"` \| `"auto"`)、`custom_instructions`。

所有 hook 输入都包含基础字段：session_id、transcript_path、cwd，以及可选的 permission_mode。

Hook 输出¶

Hook 回调返回一个同步输出对象：

# Synchronous output
SyncHookJSONOutput = TypedDict("SyncHookJSONOutput", {
    "continue_": bool,               # Whether agent should proceed (default: True)
    "stopReason": str,                # Message shown when continue_ is False
    "decision": Literal["block"],     # Block the action
    "reason": str,                    # Feedback for the agent
    "hookSpecificOutput": ...,        # Event-specific controls
}, total=False)

备注

Python SDK 使用 continue_``（带尾随下划线）以避免 Python 关键字冲突。当发送给 CLI 时，它会被自动转换回 ``continue。

工具权限¶

can_use_tool 回调允许您通过编程方式控制工具权限。

对于许多常规的工具权限检查，回调输入包含诸如 {"action": ..., "resource": ...} 之类的字段。允许/拒绝结果和可选的拒绝消息用于这些检查。updated_input 用于 SDK 路由的伪工具（例如 AskUserQuestion 和 ExitPlanMode），这些工具包含特定于工具的字段。

from typing import Any

from cortex_code_agent_sdk import (
    CortexCodeAgentOptions,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)

async def my_permission_handler(
    tool_name: str,
    tool_input: dict[str, Any],
    context: ToolPermissionContext,
) -> PermissionResultAllow | PermissionResultDeny:
    if tool_name == "Write" and str(tool_input.get("resource", "")).endswith(".env"):
        return PermissionResultDeny(message="Writing env files is not allowed")
    return PermissionResultAllow()

options = CortexCodeAgentOptions(can_use_tool=my_permission_handler)

类型

CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResult],
]

@dataclass
class ToolPermissionContext:
    signal: Any | None = None
    blocked_path: str | None = None
    decision_reason: str | None = None
    tool_use_id: str = ""
    agent_id: str | None = None

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    tool_use_id: str | None = None

@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False
    tool_use_id: str | None = None

MCP 服务器配置¶

mcp_servers 选项接受将服务器名称映射到外部 MCP 服务器配置的字典：


配置类型	描述
`McpStdioServerConfig`	通过 stdio 的外部进程。字段：`command`、`args`、`env`。
`McpSSEServerConfig`	服务器发送的事件。字段：`type="sse"`、`url`、`headers`。
`McpHttpServerConfig`	HTTP 运输。字段：`type="http"`、`url`、`headers`。

示例

options = CortexCodeAgentOptions(
    mcp_servers={
        "external": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
         },
     },
 )

错误处理¶

from cortex_code_agent_sdk import (
    query, CortexCodeAgentOptions, ResultMessage,
    CortexCodeSDKError, CLINotFoundError, CLIConnectionError,
)

try:
    async for message in query(prompt="...", options=CortexCodeAgentOptions()):
        if isinstance(message, ResultMessage) and message.is_error:
            print(f"Agent error ({message.subtype}): {message.result}")
except CLINotFoundError:
    print("Cortex CLI not found. Is it installed and on PATH?")
except CLIConnectionError:
    print("Failed to connect to CLI process")
except CortexCodeSDKError as e:
    print(f"SDK error: {e}")

错误类型¶


异常	描述
`CortexCodeSDKError`	全部 SDK 错误的基本异常
`CLINotFoundError`	在 PATH 中未找到 CLI 二进制文件
`CLIConnectionError`	无法连接或无法与 CLI 进行通信
`ProcessError`	CLI 进程异常退出
`CLIJSONDecodeError`	无法从 CLI 中解析 JSON

法律声明¶

如果 Cortex Code 配置使用模型和服务直通条款中提供的模型，则您对该模型的使用将进一步受该页面上该模型的条款的约束。

输入和输出的 Data Classification 如下表所示。


输入 Data Classification	输出 Data Classification	名称
Usage Data	客户数据	涵盖的 AI 功能 [1]

有关更多信息，请参阅 Snowflake AI 和 ML。