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

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

pip install cortex-code-agent-sdk

[dev-connection]account = "myorg-dev"user = "dev_user"authenticator = "externalbrowser" # Keeps things simple with SSOrole = "ENGINEER_ROLE"warehouse = "COMPUTE_WH"

cortex --connection my-connection --print "G'day!"

from cortex_code_agent_sdk import query, CortexCodeAgentOptions# Explicit connection nameoptions = CortexCodeAgentOptions(    connection="my-connection",    cwd=".",)# Or omit 'connection' to use default_connection_name from the toml fileoptions = CortexCodeAgentOptions(cwd=".")

import asyncioimport jsonasync 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("message", {}).get("content", []):                if block.get("type") == "text":                    print(block["text"], end="")        elif msg.get("type") == "result":            print(f"\nDone: {msg.get('subtype', 'unknown')}")    await proc.wait()asyncio.run(main())

proc = await asyncio.create_subprocess_exec(    "cortex",    "-p", "Review utils.py for bugs. Fix any issues you find.",    "--output-format", "stream-json",    "--allowed-tools", "Read",    "--allowed-tools", "Edit",    "--allowed-tools", "Glob",    "--allowed-tools", "Grep",    stdout=asyncio.subprocess.PIPE,    stderr=asyncio.subprocess.PIPE,)# ... parse NDJSON lines from proc.stdout

## Turn 1proc1 = await asyncio.create_subprocess_exec(    "cortex",    "-p", "Inspect the schema and optimise the semantic view",    "--output-format", "stream-json",    "--allowed-tools", "Read", "--allowed-tools", "Glob",    "--allowed-tools", "Grep", "--allowed-tools", "Edit",    stdout=asyncio.subprocess.PIPE,    stderr=asyncio.subprocess.PIPE,)# ... parse NDJSON lines from proc1.stdoutawait proc1.wait()## Turn 2: Launch another agent that builds on the file changes from Turn 1proc2 = await asyncio.create_subprocess_exec(    "cortex",    "-p", "Now create search services for text columns",    "--output-format", "stream-json",    "--allowed-tools", "Read", "--allowed-tools", "Write",    "--allowed-tools", "Glob", "--allowed-tools", "Grep",    stdout=asyncio.subprocess.PIPE,    stderr=asyncio.subprocess.PIPE,)# ... parse NDJSON lines from proc2.stdoutawait proc2.wait()

# buggy_calculator.pyimport jsonfrom datetime import datetimedef calculate_average(numbers):    total = 0    for num in numbers:        total += num    return total / len(numbers)  # Bug: ZeroDivisionError on empty listdef parse_user_config(config_string):    config = json.loads(config_string)    return {        "username": config["user"]["name"].strip(),    # Bug: KeyError if "user" or "name" missing        "timeout": int(config["settings"]["timeout"]), # Bug: KeyError if nested keys missing        "created": datetime.strptime(            config["metadata"]["created_at"],          # Bug: KeyError if key missing            "%Y-%m-%d"        ),    }def find_outliers(data, threshold):    mean = sum(data) / len(data)  # Bug: ZeroDivisionError on empty list    std_dev = (sum((x - mean) ** 2 for x in data) / len(data)) ** 0.5  # Same bug    return [x for x in data if abs(x - mean) > threshold * std_dev]def merge_sorted_lists(list_a, list_b):    result = []    i, j = 0, 0    while i < len(list_a) and j < len(list_b):        if list_a[i] <= list_b[j]:            result.append(list_a[i])            i += 1        else:            result.append(list_b[j])            j += 1    # Bug: remaining elements from both lists are never appended    return result

# fix_bugs.pyimport asyncioimport jsonPROMPT = """Review the file buggy_calculator.py for bugs that would cause crashes or incorrect results.For each bug you find:1. Explain what the bug is and when it would trigger2. Fix the bug in the fileBe thorough -- check for edge cases like empty inputs, missing keys, and incomplete logic."""async def main():    print("Agent starting bug analysis...\n")    proc = await asyncio.create_subprocess_exec(        "cortex",        "-p", PROMPT,        "--output-format", "stream-json",        "--allowed-tools", "Read",        "--allowed-tools", "Write",        "--allowed-tools", "Edit",        "--allowed-tools", "Glob",        "--allowed-tools", "Grep",        "--allowed-tools", "Bash",        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        msg_type = msg.get("type")        if msg_type == "assistant":            content = msg.get("message", {}).get("content", [])            for block in content:                if block.get("type") == "text":                    print(block["text"], end="")                elif block.get("type") == "tool_use":                    print(f"\n> Tool: {block['name']}")        elif msg_type == "result":            subtype = msg.get("subtype", "unknown")            duration = msg.get("duration_ms", 0)            if subtype == "success":                print(f"\n\nAll bugs fixed successfully. ({duration}ms)")            else:                print(f"\n\nAgent finished with status: {subtype} ({duration}ms)")    await proc.wait()asyncio.run(main())

Agent starting bug analysis...I'll review buggy_calculator.py for bugs that could cause crashes or incorrect results.> Tool: ReadI've read the file. Let me analyse each function for bugs.**Bug 1: `calculate_average` -- ZeroDivisionError on empty list**The function divides by `len(numbers)` without checking if the list is empty.When called with `calculate_average([])`, this crashes with `ZeroDivisionError`.> Tool: EditFixed: Added an early return for empty lists:    if not numbers:        return 0**Bug 2: `parse_user_config` -- KeyError on missing nested keys**The function accesses `config["user"]["name"]`, `config["settings"]["timeout"]`,and `config["metadata"]["created_at"]` directly. If any of these keys are missing,it raises `KeyError`.> Tool: EditFixed: Replaced direct key access with `.get()` and sensible defaults:    user = config.get("user", {})    settings = config.get("settings", {})    metadata = config.get("metadata", {})**Bug 3: `find_outliers` -- ZeroDivisionError on empty list**Same pattern as `calculate_average` -- `sum(data) / len(data)` crashes on empty input.> Tool: EditFixed: Added an early return for empty data lists.**Bug 4: `merge_sorted_lists` -- remaining elements never appended**After the `while` loop exits, any remaining elements in `list_a` or `list_b` aresilently dropped. For example, `merge_sorted_lists([1, 3, 5], [2, 4])` returns`[1, 2, 3, 4]` instead of `[1, 2, 3, 4, 5]`.> Tool: EditFixed: Added `result.extend(list_a[i:])` and `result.extend(list_b[j:])` after the loop.All bugs fixed successfully. (12850ms)

customer_id,tenure_months,monthly_charges,total_charges,contract_type,payment_method,num_support_tickets,has_online_security,has_tech_support,churnedC001,12,64.50,774.00,month-to-month,credit_card,3,0,0,1C002,48,89.20,4281.60,one_year,bank_transfer,0,1,1,0C003,3,29.99,89.97,month-to-month,credit_card,5,0,0,1C004,72,105.00,7560.00,two_year,auto_pay,1,1,1,0C005,1,45.00,45.00,month-to-month,credit_card,2,0,0,1...

# ml_feature_pipeline.pyimport asyncioimport json# Tools the agent is allowed to use across all turns.ALLOWED_TOOLS = ["Read", "Write", "Edit", "Glob", "Grep", "Bash"]# JSON Schema for the final structured output -- Turn 4's response# must conform to this shape.FEATURE_SCHEMA = {    "type": "object",    "properties": {        "target_column": {"type": "string"},        "selected_features": {            "type": "array",            "items": {                "type": "object",                "properties": {                    "name": {"type": "string"},                    "importance_score": {"type": "number"},                    "category": {                        "type": "string",                        "enum": ["original", "engineered"],                    },                    "rationale": {"type": "string"},                },                "required": ["name", "importance_score", "category", "rationale"],            },        },        "dropped_features": {            "type": "array",            "items": {                "type": "object",                "properties": {                    "name": {"type": "string"},                    "reason": {"type": "string"},                },                "required": ["name", "reason"],            },        },    },    "required": ["target_column", "selected_features", "dropped_features"],}async def run_agent_turn(prompt: str, output_schema: dict | None = None) -> dict:    """Run a single agent turn via the cortex CLI and return parsed results.    Returns a dict with:        - "text": concatenated assistant text blocks        - "result_subtype": "success" | "error" | ...        - "duration_ms": execution time        - "structured_output": parsed JSON if output_schema was provided, else None    """    cmd = ["cortex", "-p", prompt, "--output-format", "stream-json"]    for tool in ALLOWED_TOOLS:        cmd.extend(["--allowed-tools", tool])    if output_schema:        cmd.extend(["--output-format-json-schema", json.dumps(output_schema)])    proc = await asyncio.create_subprocess_exec(        *cmd,        stdout=asyncio.subprocess.PIPE,        stderr=asyncio.subprocess.PIPE,    )    text_parts = []    result_subtype = "unknown"    duration_ms = 0    structured_output = None    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        msg_type = msg.get("type")        if msg_type == "assistant":            content = msg.get("message", {}).get("content", [])            for block in content:                if block.get("type") == "text":                    text_parts.append(block["text"])                    print(block["text"], end="")                elif block.get("type") == "tool_use":                    print(f"\n> Tool: {block['name']}")        elif msg_type == "result":            result_subtype = msg.get("subtype", "unknown")            duration_ms = msg.get("duration_ms", 0)
本文来源：InfoQ