如何在Snowflake上构建简易AI Agent？Cortex Code Agent SDK深度解析｜技术实践

2026 年，智能体将在企业级应用中取得哪些实质性突破？ 点击下载 "《2026 年 AI 与数据发展预测》白皮书，获悉专家一手前瞻，抢先拥抱新的工作方式！ 将 Cortex Code 从交互式 CLI 转变为一个可编程引擎，用于运行自主 AI 工作流。这些工作流既可以在你的本地机器上运行，也可以在 Snowflake 内部以服务端方式运行。 从“它在聊天里能用”到“它在生产环境里能用”的跨越，通常就是事情开始崩塌的地方。聊天适合一次性的代码片段，但它不是一种自动化策略。一旦你需要编排复杂任务，比如在服务端工作流中运行 Agent，或在 CI/CD 中运行 Agent，你需要的是一个可编程引擎，而不是一个文本框。 Cortex Code Agent SDK（或称 “CoCo” SDK）就是 Snowflake 对此给出的答案。它采用与 CLI 中相同的 Agentic 引擎，并将其封装成一个可编程接口。不是由你来驱动 Agent，而是由你的代码来驱动。如果你一直想启动一个真正能在服务端工作流中执行任务的 AI 助手，那么这就是你一直在寻找的工具包。 我们到底在说什么？ 如果你最近一直在使用 Snowflake，你很可能已经见过 Cortex Code。它是 Snowflake 的 CLI Agent，不只是“建议”代码——它会真正围绕任务进行推理，读取你的本地文件，运行 SQL，并不断迭代，直到达成目标。 在底层，它正在调用 Claude Sonnet 4.6 或 OpenAI GPT-5.x 这类高端模型。SDK 只是为你提供了一种以编程方式接入这一循环的方法。 CLI 与 SDK：真正的区别 CLI 是给你用的：你坐在桌前，实时看着终端中的循环过程；SDK 是适用于凌晨 2 点运行的 Python 脚本：需要检查 100 张表，优化一个视图，并在无需你盯着的情况下生成一份报告。 为什么要使用 SDK？ 自动化是显而易见的答案，但真正的“为什么”归结为控制力： 可重复性：在一个庞大的代码库中运行同一个 Agentic 审计，而不必反复手动操作到手腕酸痛；条件逻辑：你的 Python 代码可以为 Agent 设置“关卡”。例如：如果安全扫描发现高风险漏洞，就调用 Agent 来修复；如果没有，就直接继续；结构化输出：你可以强制 Agent 返回类型化 JSON，而不是让它给你一段闲聊式的文字，这样你的下游代码才能真正使用这些数据；自定义工具（MCP）：你可以使用 Model Context Protocol 将 Agent 接入你的内部 API，例如 Jira 或 Slack。 SDK 将 Cortex Code 从一个人在环路中的助手，转变为一个由你的 Python 或 TypeScript 代码编排的可编程 Agentic 引擎。 使用场景 以下是 SDK 大放异彩的一些具体场景： 自动化代码审查：扫描每个 PR 中的漏洞、安全问题或风格违规，并返回结构化发现；数据管道验证：在 ETL 运行后，让 Agent 检查表结构、验证数据质量，并生成报告；语义视图优化：一个多轮管道，可检查 Snowflake 表，优化语义视图，创建搜索服务，并自主完成整合；代码库迁移：分析文件中的弃用模式，并在整个代码仓库中应用修复；机器学习中的特征工程：以 Agentic 方式探索数据集，生成候选特征，评估它们，并为模型选择最佳特征子集；事件响应自动化：当告警触发时，让 Agent 读取日志、追踪问题，并提出修复方案。 准备基础环境 安装 CLI 和 SDK 你需要安装 CLI 和 SDK 包。安装 Cortex Code CLI： curl -LsS https://ai.snowflake.com/static/cc-scripts/install.sh | sh 从 PyPI安装 SDK（Python ≥ 3.10）： pip install cortex-code-agent-sdk 建立 Snowflake 连接 SDK 会复用你现有的 ~/.snowflake/connections.toml。这是一个很大的优势，因为你不需要管理一套新的凭据。 第 1 步：配置你的连接文件 创建或编辑 ~/.snowflake/connections.toml： [dev-connection] account = "myorg-dev" user = "dev_user" authenticator = "externalbrowser" # Keeps things simple with SSO role = "ENGINEER_ROLE" warehouse = "COMPUTE_WH" 第 2 步：验证你的连接是否可用 cortex --connection my-connection --print "G'day!" 第 3 步：将连接传递给 SDK from cortex_code_agent_sdk import query, CortexCodeAgentOptions # Explicit connection name options = CortexCodeAgentOptions( connection="my-connection", cwd=".", ) # Or omit 'connection' to use default_connection_name from the toml file options = CortexCodeAgentOptions(cwd=".") connection选项会直接映射到 CLI 的 --connection 标志。SDK 会生成 CLI 子进程，该子进程会使用指定的连接配置文件或默认连接配置文件进行身份验证。 最小可运行示例 import asyncio import json async def main(): proc = await asyncio.create_subprocess_exec( "cortex", "-p", "What files are in this directory?", "--output-format", "stream-json", "--allowed-tools", "Read", "--allowed-tools", "Glob", "--allowed-tools", "Grep", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) while True: line = await proc.stdout.readline() if not line: break text = line.decode().strip() if not text: continue try: msg = json.loads(text) except json.JSONDecodeError: continue if msg.get("type") == "assistant": for block in msg.get(