Cortex Code Agent SDK 快速入门¶

本主题将带您构建一个 AI 代理，该代理可读取数据管道脚本、发现其中的错误，并使用 Cortex Code Agent SDK 自动修复这些错误。

您将完成以下内容：

使用 Cortex Code Agent SDK 设置项目。
创建一个包含若干错误的数据管道脚本。
运行代理，使其无需人工干预即可自动发现并修复错误。

先决条件¶

**Node.js 18+**（适用于 TypeScript）或 **Python 3.10+**（适用于 Python）。
Snowflake 连接 通过 Snowflake CLI 连接设置进行配置，通常存放在 ~/.snowflake/connections.toml 中。对于已有配置的环境，也可以继续使用 ~/.snowflake/config.toml`（请参阅 :doc:`配置连接 </developer-guide/snowflake-cli/connecting/configure-connections>）：
```
[my-connection]
account = "myorg-myaccount"
user = "myuser"
authenticator = "externalbrowser"
```

设置¶

1.安装 Cortex Code CLI¶

安装 CLI：

curl -LsS https://ai.snowflake.com/static/cc-scripts/install.sh | sh

验证安装：

cortex --version

2.设置项目¶

创建并进入项目目录：

mkdir my-agent && cd my-agent

3.安装 SDK¶

npm init -y
npm install cortex-code-agent-sdk

python3 -m venv .venv && source .venv/bin/activate
pip install cortex-code-agent-sdk

创建数据管道脚本¶

创建一个包含刻意错误的数据管道脚本，供代理修复：

import json

def load_results(rows):
    """Load query results into a list of campaign dicts."""
    return [
        {
            "campaign": row["campaign_name"],
            "impressions": row["impressions"],
            "clicks": row["clicks"],
            "conversions": row["conversions"],
        }
        for row in rows
    ]

def compute_conversion_rate(results):
    """Add conversion_rate (conversions / clicks) to each campaign."""
    for row in results:
        row["conversion_rate"] = row["conversions"] / row["clicks"]  # Bug: ZeroDivisionError when clicks is 0
    return results

def format_report(results):
    """Return a JSON summary with total conversions and the top campaign."""
    total = sum(r["conversions"] for r in results)
    top = max(results, key=lambda r: r["conversion_rate"])  # Bug: crashes on empty list
    return json.dumps({"total_conversions": total, "top_campaign": top["campaign"]})

interface Row {
  campaign_name: string;
  impressions: number;
  clicks: number;
  conversions: number;
}

interface Result {
  campaign: string;
  impressions: number;
  clicks: number;
  conversions: number;
  conversion_rate?: number;
}

export function loadResults(rows: Row[]): Result[] {
  return rows.map(row => ({
    campaign: row.campaign_name,
    impressions: row.impressions,
    clicks: row.clicks,
    conversions: row.conversions,
  }));
}

export function computeConversionRate(results: Result[]): Result[] {
  return results.map(row => ({
    ...row,
    conversion_rate: row.conversions / row.clicks,  // Bug: NaN or Infinity when clicks is 0
  }));
}

export function formatReport(results: Result[]): string {
  const total = results.reduce((sum, r) => sum + r.conversions, 0);
  const top = results.reduce((best, r) =>
    r.conversion_rate! > best.conversion_rate! ? r : best  // Bug: crashes on empty array
  );
  return JSON.stringify({ total_conversions: total, top_campaign: top.campaign });
}

此代码存在两个问题：

computeConversionRate / compute_conversion_rate 在未检查是否为零的情况下除以 clicks，对于没有点击的活动，会返回 NaN 或 Infinity (TypeScript)，或在 Python 中抛出 ZeroDivisionError。
formatReport / format_report 在未检查结果列表是否为空的情况下调用 max / reduce，当没有数据行时，会在 Python 中抛出 ValueError 或 TypeError (TypeScript)。

构建一个可自动发现并修复错误的代理¶

// agent.mjs
import { query } from "cortex-code-agent-sdk";

// Agentic loop: streams messages as the agent works
for await (const message of query({
  prompt: "Review report.ts for bugs in the data pipeline. Fix any issues you find.",
  options: {
    cwd: process.cwd(),
    connection: "my-connection",          // Snowflake CLI connection name
    allowedTools: ["Read", "Edit", "Bash"],  // Auto-approve these tools without prompting
  },
})) {
  // Print human-readable output
  if (message.type === "assistant") {
    for (const block of message.content) {
      if (block.type === "text") {
        process.stdout.write(block.text);  // Agent's reasoning
      } else if (block.type === "tool_use") {
        console.log(`Tool: ${block.name}`);  // Tool being called
      }
    }
  } else if (message.type === "result") {
    console.log(`\nDone: ${message.subtype}`);  // Final result
  }
}

# agent.py
import asyncio
from cortex_code_agent_sdk import query, AssistantMessage, ResultMessage, CortexCodeAgentOptions

async def main():
    # Agentic loop: streams messages as the agent works
    async for message in query(
        prompt="Review report.py for bugs in the data pipeline. Fix any issues you find.",
        options=CortexCodeAgentOptions(
            cwd=".",
            connection="my-connection",              # Snowflake CLI connection name
            allowed_tools=["Read", "Edit", "Bash"],  # Auto-approve these tools without prompting
        ),
    ):
        # Print human-readable output
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(block.text, end="")  # Agent's reasoning
                elif hasattr(block, "name"):
                    print(f"Tool: {block.name}")  # Tool being called
        elif isinstance(message, ResultMessage):
            print(f"\nDone: {message.subtype}")  # Final result

asyncio.run(main())

该代码包含三个主要部分：

query()：创建代理循环的主入口函数。它会返回一个异步迭代器，您可以在对应语言的异步循环中消费该迭代器，以在代理执行过程中实时获取消息流。完整的 API 内容请参考 TypeScript 或 Python 参考。
prompt：您希望代理执行的任务。用于告知代理需要完成的具体工作。
options：代理配置项。connection 用于指定用于身份验证的 Snowflake CLI 连接；allowedTools 指定哪些工具可自动批准（无需提示）；disallowedTools 可用于完全阻止某些工具。其他选项还包括 model、mcp_servers 等。

流式循环在代理思考、调用工具、观察结果并决定下一步操作时持续运行。每次迭代都会输出一条消息：包括代理推理、工具调用、工具结果或最终结果。SDK 负责整体编排。

运行代理¶

node agent.mjs

python3 agent.py

运行完成后，请检查报告文件。您将看到防御性代码，用于处理空结果和零点击活动等情况。您的代理可自主执行以下操作：

读取文件以理解代码。
分析逻辑并识别可能导致崩溃的边界情况。
编辑文件并加入合理的错误处理机制。

多轮对话¶

在需要共享上下文并进行交互会话的场景中，请使用 客户端 API：

import {
  createCortexCodeSession,
  type CortexCodeEvent,
} from "cortex-code-agent-sdk";

async function printResponse(stream: AsyncIterable<CortexCodeEvent>) {
  for await (const event of stream) {
    if (event.type === "assistant") {
      for (const block of event.content) {
        if (block.type === "text") process.stdout.write(block.text);
      }
    } else if (event.type === "result") {
      break;
    }
  }
}

const session = await createCortexCodeSession({
  cwd: process.cwd(),
  connection: "my-connection",
});

// First turn
await session.send("Summarize what report.ts does and what data it expects.");
await printResponse(session.stream());

// Second turn (same session, remembers context)
await session.send("Now add type annotations to each function.");
await printResponse(session.stream());

await session.close();

from cortex_code_agent_sdk import CortexCodeSDKClient, CortexCodeAgentOptions, AssistantMessage

async with CortexCodeSDKClient(
    CortexCodeAgentOptions(
        cwd=".",
        connection="my-connection",
    )
) as client:
    # First turn
    await client.query("Summarize what report.py does and what data it expects.")
    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 (same session, remembers context)
    await client.query("Now add docstrings to each 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="")

尝试其他提示¶

代理配置完成后，您可以尝试不同的提示：

"Add comprehensive type hints to all functions in report.py"
"Write a SQL query that finds the top 10 campaigns by conversion rate"
"Add input validation to all functions in report.py"
"Create a README.md documenting the functions in report.py"

关键概念¶

权限模式¶

权限模式用于控制工具调用时的人类干预级别：


模式	行为	用例
``"bypassPermissions"``（带安全标志）	执行所有工具调用而无需提示确认。需要 `allowDangerouslySkipPermissions: true` (TypeScript) 或 `allow_dangerously_skip_permissions=True` (Python)。	沙盒化 CI，完全可信环境
`"default"`	使用标准权限检查。在 SDK 会话中，可通过配置 `allowedTools`、`disallowedTools` 或 `canUseTool` 来控制需要权限校验的工具。	使用显式权限策略的受控工作流
`"autoAcceptPlans"`	自动批准计划请求及计划退出确认。它不会绕过常规工具权限。	适用于需要自动通过计划审批的特定工作流
`"plan"`	从规划开始；批准 `ExitPlanMode` 后可继续执行，之后恢复正常权限模式	代码审查、分析

要对单个工具调用进行精细控制，请使用 canUseTool 回调。请参阅处理审批与用户输入了解详细信息。

后续步骤¶

处理审批与用户输入：通过 canUseTool 回调控制代理可使用的工具。
TypeScript SDK 参考：query()、createCortexCodeSession()、类型及事件的完整 API 文档。
Python SDK 参考：query()、CortexCodeSDKClient、MCP、工具及 Hook 的完整 API 文档。

法律声明¶

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

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


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

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