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)            structured_output = msg.get("structured_output")            print(f"\n[Turn complete: {result_subtype} ({duration_ms}ms)]")    await proc.wait()    return {        "text": "".join(text_parts),        "result_subtype": result_subtype,        "duration_ms": duration_ms,        "structured_output": structured_output,    }async def main():        ## Turn 1: Explore the dataset    print("=" * 60)    print("TURN 1: Data Exploration")    print("=" * 60)    turn1 = await run_agent_turn("""        Read customer_churn.csv and profile it:        - List every column with its dtype, unique count, and missing-value count        - For numeric columns, report min/median/max and any outliers        - For categorical columns, report value distributions        - Compute correlations with the 'churned' target column        - Summarise your findings at the end    """)    # Decision gate: check if the data is usable     if "missing" in turn1["text"].lower() and "100%" in turn1["text"]:        print("\n[PIPELINE] Dataset has entirely empty columns. Aborting.")        return        ## Turn 2: Engineer candidate features     print("\n" + "=" * 60)    print("TURN 2: Feature Engineering")    print("=" * 60)    await run_agent_turn("""        Based on the customer_churn.csv dataset, write a Python script called        feature_engineering.py that:        1. Reads customer_churn.csv        2. Generates candidate features:           - charge_per_month = total_charges / max(tenure_months, 1)           - ticket_rate = num_support_tickets / max(tenure_months, 1)           - has_protection = has_online_security | has_tech_support           - is_new_customer = 1 if tenure_months <= 6 else 0           - One-hot encode contract_type and payment_method        3. Drops customer_id (non-predictive) and the original           categorical columns after encoding        4. Saves the result to features_output.csv        5. Prints the shape and column list when done        Make sure the script handles edge cases (division by zero,        missing values) and runs without errors.    """)    ## Turn 3: Evaluate feature importance    print("\n" + "=" * 60)    print("TURN 3: Feature Evaluation")    print("=" * 60)    turn3 = await run_agent_turn("""        Run feature_engineering.py to generate features_output.csv.        Then evaluate each feature's predictive power for the 'churned' target:        1. Compute the absolute Pearson correlation of every feature with 'churned'        2. Rank features from most to least important        3. Flag any features with near-zero variance or high mutual correlation           (>0.9 with another feature)        Print a ranked table of features with their scores.    """)    ## Turn 4: Structured selection    print("\n" + "=" * 60)    print("TURN 4: Feature Selection (Structured Output)")    print("=" * 60)    turn4 = await run_agent_turn(        f"""        Here are the feature evaluation results from a previous analysis:        {turn3['text']}        Based on these results, select the final feature set for a churn        prediction model. Include features with importance score > 0.05.        Drop features that are redundant or have near-zero variance.        Return your selection as structured JSON matching the provided schema.        For each selected feature, include its importance score, whether it's        an original or engineered feature, and a brief rationale.        For each dropped feature, explain why it was excluded.        """,        output_schema=FEATURE_SCHEMA,    )    ## Use the structured result programmatically    result = turn4["structured_output"]    if result:        print("\n" + "=" * 60)        print("PIPELINE RESULT")        print("=" * 60)        print(f"Target: {result['target_column']}")        print(f"Selected features: {len(result['selected_features'])}")        for f in result["selected_features"]:            print(f"  {f['importance_score']:.3f}  {f['name']:30s}  [{f['category']}]")        print(f"\nDropped features: {len(result['dropped_features'])}")        for f in result["dropped_features"]:            print(f"  x {f['name']:30s}  ({f['reason']})")        # Write the final feature manifest for downstream consumers        with open("feature_manifest.json", "w") as fp:            json.dump(result, fp, indent=2)        print("\nFeature manifest saved to feature_manifest.json")asyncio.run(main())

============================================================TURN 1: Data Exploration============================================================I'll read customer_churn.csv and profile the dataset.> Tool: ReadThe dataset has 50 rows and 10 columns. Here's the profile:| Column               | Dtype   | Unique | Missing | Min    | Median  | Max      ||----------------------|---------|--------|---------|--------|---------|----------|| customer_id          | string  | 50     | 0       | -      | -       | -        || tenure_months        | int     | 38     | 0       | 1      | 20.0    | 72       || monthly_charges      | float   | 46     | 0       | 29.99  | 74.0    | 110.0    || total_charges        | float   | 50     | 0       | 30.0   | 1976.0  | 7560.0   || contract_type        | string  | 3      | 0       | -      | -       | -        || payment_method       | string  | 3      | 0       | -      | -       | -        || num_support_tickets  | int     | 8      | 0       | 0      | 1.0     | 7        || has_online_security  | int     | 2      | 0       | 0      | -       | 1        || has_tech_support     | int     | 2      | 0       | 0      | -       | 1        || churned              | int     | 2      | 0       | 0      | -       | 1        |**Correlations with `churned`**:- tenure_months:       -0.82 (strong negative -- longer tenure = less churn)- num_support_tickets:  0.79 (strong positive -- more tickets = more churn)- has_online_security: -0.72- has_tech_support:    -0.60- monthly_charges:     -0.48- total_charges:       -0.74No columns are entirely missing. The dataset is clean and usable.[Turn complete: success (14520ms)]============================================================TURN 2: Feature Engineering============================================================I'll create feature_engineering.py with the candidate features.> Tool: WriteI've written feature_engineering.py. The script:1. Reads customer_churn.csv2. Engineers 4 new features: charge_per_month, ticket_rate, has_protection, is_new_customer3. One-hot encodes contract_type (3 categories) and payment_method (3 categories)4. Drops customer_id and the original categorical columns> Tool: BashOutput shape: (50, 16). Columns:tenure_months, monthly_charges, total_charges, num_support_tickets,has_online_security, has_tech_support, churned, charge_per_month,ticket_rate, has_protection, is_new_customer, contract_month-to-month,contract_one_year, contract_two_year, payment_auto_pay, payment_bank_transfer[Turn complete: success (21340ms)]============================================================TURN 3: Feature Evaluation============================================================I'll run feature_engineering.py and evaluate feature importance.> Tool: BashFeature Importance (absolute Pearson correlation with `churned`):| Rank | Feature                   | Correlation ||------|---------------------------|-------|| 1    | tenure_months             | 0.821 || 2    | contract_month-to-month   | 0.806 || 3    | num_support_tickets       | 0.789 || 4    | total_charges             | 0.742 || 5    | is_new_customer           | 0.723 || 6    | has_online_security       | 0.718 || 7    | ticket_rate               | 0.695 || 8    | has_tech_support          | 0.601 || 9    | has_protection            | 0.584 || 10   | monthly_charges           | 0.483 || 11   | charge_per_month          | 0.412 || 12   | contract_two_year         | 0.378 || 13   | payment_auto_pay          | 0.305 || 14   | contract_one_year         | 0.267 || 15   | payment_bank_transfer     | 0.041 |**Redundancy flags**:- total_charges & tenure_months: correlation 0.94 (>0.9 threshold)- has_protection & has_online_security: correlation 0.91 (>0.9 threshold)[Turn complete: success (18760ms)]============================================================TURN 4: Feature Selection (Structured Output)============================================================Based on the evaluation results, I'll select the optimal feature setand return structured JSON.[Turn complete: success (9830ms)]============================================================PIPELINE RESULT============================================================Target: churnedSelected features: 9  0.821  tenure_months                 [original]  0.806  contract_month-to-month       [engineered]  0.789  num_support_tickets           [original]  0.723  is_new_customer               [engineered]  0.718  has_online_security           [original]  0.695  ticket_rate                   [engineered]  0.601  has_tech_support              [original]  0.483  monthly_charges               [original]  0.412  charge_per_month              [engineered]Dropped features: 6  x total_charges                   (redundant with tenure_months, r=0.94)  x has_protection                  (redundant with has_online_security, r=0.91)  x contract_two_year               (below importance threshold after multicollinearity check)  x payment_auto_pay                (moderate importance but highly correlated with contract_type features)  x contract_one_year               (below importance threshold after multicollinearity check)  x payment_bank_transfer           (near-zero correlation: 0.041)Feature manifest saved to feature_manifest.json

# Development: full autonomy via CLIdev_cmd = [    "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",]# Production: scoped permissions via SDKprod = CortexCodeAgentOptions(    cwd="/path/to/project",    model="claude-opus-4-6",    allowed_tools=["Read", "Glob", "Grep", "snowflake_sql_execute"],    disallowed_tools=["Bash", "Write"],    can_use_tool=my_permission_callback,)

# Multi-turn: each query() call builds on the previous context# Turn 1async with CortexCodeSDKClient(options) as client:    await client.query("Read the database schema and summarise the tables.")    async for msg in client.receive_response():        handle_message(msg)# Turn 2 knows about the schema from Turn 1    await client.query("Now create a semantic view for the customer table.")    async for msg in client.receive_response():        handle_message(msg)

from cortex_code_agent_sdk import PermissionResultAllow, PermissionResultDenyasync def can_use_tool(tool_name, tool_input, context):    # Allow read-only tools unconditionally    if tool_name in ("Read", "Glob", "Grep"):        return PermissionResultAllow()    # Block destructive commands    if tool_name == "Bash" and "rm " in tool_input.get("command", ""):        return PermissionResultDeny(message="Destructive commands not allowed")    # Allow but add a timeout constraint    if tool_name == "Bash":        return PermissionResultAllow(            updated_input={**tool_input, "command": f"timeout 30 {tool_input['command']}"}        )    return PermissionResultDeny(message=f"Tool '{tool_name}' requires approval")

options = CortexCodeAgentOptions(    allowed_tools=["Read", "Write", "Edit", "Glob", "Grep", "Bash"],    hooks={        "PreToolUse": my_pre_tool_logger,        "PostToolUse": my_post_tool_logger,        "Stop": my_completion_handler,    },)

from cortex_code_agent_sdk import tool, create_sdk_mcp_server@tool("search_jira", "Search Jira tickets", {"query": str, "project": str})async def search_jira(args):    results = await jira_client.search(args["query"], project=args["project"])    return {"content": [{"type": "text", "text": json.dumps(results)}]}server = create_sdk_mcp_server(name="internal", version="1.0.0", tools=[search_jira])options = CortexCodeAgentOptions(    allowed_tools=["Read", "Write", "Edit", "Glob", "Grep", "Bash"],    mcp_servers={"internal": server},)