Schnellstart für Cortex Code Agent-SDK¶

Voraussetzungen¶

• Node.js 22+ (für TypeScript) oder Python 3.10+ (für Python).

• Snowflake-Verbindung, über Snowflake-CLI-Verbindungseinstellungen konfiguriert – normalerweise in:file:~/.snowflake/connections.toml mit ~/.snowflake/config.toml; wird auch für bestehende Setups unterstützt (siehe Verbindungen konfigurieren):

  CopyExpand

  [my-connection]
  account = "myorg-myaccount"
  user = "myuser"
  authenticator = "externalbrowser"
  

  Show lessSee more

  Scroll to top

Setup¶

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

cortex --version

mkdir my-agent && cd my-agent

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

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

Erstellen eines Datenpipeline-Skripts¶

Erstellen eines Datenpipeline-Skripts mit einigen absichtlichen Fehlern, die der Agent beheben soll:

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 });
}

Dieser Code hat zwei Fehler:

1. computeConversionRate / compute_conversion_rate dividiert durch clicks, ohne auf null zu prüfen. Dadurch wird bei Kampagnen ohne Klicks NaN oder Infinity (TypeScript) zurückgegeben oder ein ZeroDivisionError (Python) ausgelöst.

2. formatReport / format_report ruft max / reduce für die Ergebnisliste auf, ohne zu prüfen, ob diese leer ist. Dadurch wird ein ValueError (Python) oder TypeError (TypeScript) ausgelöst, wenn keine Zeilen vorhanden sind.

Erstellen eines Agenten, der Fehler findet und behebt¶

// 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())

Dieser Code besteht aus drei Hauptteilen:

1. Abfrage: Der Haupteinstiegspunkt, der die Agentenschleife erzeugt. Sie gibt einen asynchronen Iterator zurück, den Sie in der Syntax der asynchronen Schleife Ihrer Sprache verwenden, um Nachrichten zu streamen, während der Agent funktioniert. Vollständige Informationen zur API sehen Sie in der Referenz TypeScript oder Python.

2. Aufforderung: Was der Agent tun soll. Teilt dem Agenten mit, welche Aufgabe er erledigen soll.

3. Optionen: Konfiguration für den Agenten. connection gibt an, mit welcher Snowflake-CLI-Verbindung authentifiziert werden soll. allowedTools legt fest, welche Tools ohne Aufforderung automatisch genehmigt werden, und disallowedTools kann Tools vollständig blockieren. Weitere Optionen sind model, mcp_servers und mehr.

Die Streaming-Schleife wird ausgeführt, während der Agent denkt, Tools aufruft, die Ergebnisse beobachtet und entscheidet, was als Nächstes zu tun ist. Jede Iteration ergibt eine Nachricht: die Argumentation des Agenten, einen Aufruf des Tools, ein Ergebnis des Tools oder das Endergebnis. Das SDK übernimmt die Orchestrierung.

node agent.mjs

python3 agent.py

Multi-Turn-Konversationen¶

Für interaktive Sitzungen, bei denen Sie mehrere Eingabeaufforderungen mit gemeinsamem Kontext senden, verwenden Sie die Client-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="")

Die wichtigsten Konzepte¶

Nächste Schritte¶

• Verarbeitung von Genehmigungen und Benutzereingaben: Steuern Sie mit dem Callback canUseTool, welche Tools der Agent verwenden darf.

• TypeScript SDK-Referenz: Vollständige API-Dokumentation für query(), createCortexCodeSession(), Typen und Ereignisse.

• Python-SDK-Referenz: Vollständige API-Dokumente für query(),``CortexCodeSDKClient``, MCP–Tools und -Hooks.

Rechtliche Hinweise¶

Wenn Ihre Cortex Code-Konfiguration ein Modell verwendet, das im Rahmen der Modell- und Service-Pass-Through-Bedingungen bereitgestellt wurde, unterliegt Ihre Nutzung dieses Modells zusätzlich den Bedingungen für dieses Modell auf dieser Seite.

Die Datenklassifizierung der Eingaben und Ausgaben ist in der folgenden Tabelle aufgeführt.

Weitere Informationen dazu finden Sie unter KI und ML in Snowflake.