Démarrage rapide du SDK Agent Cortex Code¶

Conditions préalables¶

• Node.js 22+ (pour TypeScript) ou Python 3.10+ (pour Python).

• Connexion Snowflake configurée via les paramètres de connexion de la CLI Snowflake, généralement dans ~/.snowflake/connections.toml, avec ~/.snowflake/config.toml également pris en charge pour les configurations existantes (voir Configuration des connexions) :

  CopyExpand

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

  Show lessSee more

  Scroll to top

Configuration¶

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

Créer un script de pipeline de données¶

Créez un script de pipeline de données contenant des bogues intentionnels à corriger par l’agent :

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

Ce code pose deux problèmes :

1. computeConversionRate/compute_conversion_rate divise par clicks sans vérifier la présence de zéro, retournant NaN ou Infinity (TypeScript) ou générant une ZeroDivisionError (Python) pour les campagnes sans clics.

2. formatReport/format_report appelle max/reduce sur la liste de résultats sans vérifier s’il est vide, ce qui génère une ValueError (Python) ou TypeError (TypeScript) lorsqu’il n’y a pas de lignes.

Créez un agent qui trouve et corrige les bogues¶

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

Ce code comporte trois parties principales :

1. query(): Le point d’entrée principal qui crée la boucle agentique. Il renvoie un itérateur asynchrone que vous pouvez utiliser dans la syntaxe de boucle asynchrone de votre langage pour lire les messages au fur et à mesure que l’agent travaille. Voir l’ensemble des API dans la référence TypeScript ou Python.

2. invite : Ce que vous voulez que fasse l’agent. Il indique à l’agent quelle tâche effectuer.

3. options Configuration de l’agent. connection spécifie quelle connexion CLI Snowflake avec laquelle s’authentifier. allowedTools spécifie quels outils sont approuvés automatiquement sans invite, et``disallowedTools`` peut bloquer complètement des outils. Les autres options incluent model, mcp_servers et d’autres encore.

La boucle de streaming s’exécute au fur et à mesure que l’agent réfléchit, appelle les outils, observe les résultats et décide des prochaines actions. Chaque itération produit un message : le raisonnement de l’agent, un appel d’outil, un résultat d’outil ou le résultat final. Le SDK gère l’orchestration.

node agent.mjs

python3 agent.py

Conversation à plusieurs tours¶

Pour les sessions interactives où vous envoyez plusieurs invites avec un contexte partagé, utilisez l’API client :

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="")

Concepts clés¶

Prochaines étapes¶

• Gérer les approbations et les entrées utilisateur : Contrôlez les outils que l’agent peut utiliser avec le rappel canUseTool.

• Référence SDK TypeScript : Terminez les documents API pour query() ,``createCortexCodeSession()``, les types et les événements.

• Référence de SDK Python : Terminez les documents API pour query() ,``CortexCodeSDKClient``, les outils MCP et les hooks.

Avis juridiques¶

Lorsque votre configuration de Cortex Code utilise un modèle fourni conformément aux Conditions de répercussion relatives aux modèles et aux services, votre utilisation de ce modèle est en outre soumise aux conditions de ce modèle sur cette page.

La classification des données d’entrées et de sorties est présentée dans la table suivante.

Pour plus d’informations, reportez-vous à Snowflake AI et ML.