Référence de SDK d’agent Cortex Code – Python

Cette rubrique fournit la référence d’API complète pour le SDK d’Agent Cortex Code pour Python, y compris l’ensemble des fonctions, classes et types.

Installation

pip install cortex-code-agent-sdk

Requiert Python 3.10 ou une version ultérieure. Dépendances : anyio, mcp, typing_extensions. Importer le paquet en tant que cortex_code_agent_sdk. Le SDK attend la CLI Cortex Code à installer séparément. Si ce n’est pas sur votre PATH, définir CORTEX_CODE_CLI_PATH=/path/to/cortex ou transmettre cli_path dans CortexCodeAgentOptions.

Fonctions

query()

La fonction principale pour interagir avec Cortex Code. Renvoie un itérateur asynchrone qui produit les messages au fur et à mesure qu’ils arrivent.

async def query(
    *,
    prompt: str | AsyncIterable[dict],
    options: CortexCodeAgentOptions | None = None,
    transport: Transport | None = None,
) -> AsyncIterator[Message]: ...

Paramètres

Paramètre

Type

Description

prompt

str | AsyncIterable[dict]

Chaîne d’invite utilisateur, ou itérable asynchrone de dicts de messages pour l’entrée de streaming

options

CortexCodeAgentOptions | None

Options de configuration. La valeur par défaut est CortexCodeAgentOptions()

transport

Transport | None

Transport personnalisé. La valeur par défaut est un transport CLI de sous-processus

Renvoie

Un itérateur asynchrone produisant des objets Message (AssistantMessage, ResultMessage, UserMessage, SystemMessage, StreamEvent).

Exemple

import asyncio
from cortex_code_agent_sdk import query, AssistantMessage, ResultMessage, CortexCodeAgentOptions

async def main():
    async for message in query(
        prompt="Fix the bug in utils.py",
        options=CortexCodeAgentOptions(cwd="/path/to/project"),
    ):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if hasattr(block, "text"):
                    print(block.text, end="")
        elif isinstance(message, ResultMessage):
            print(f"\nDone: {message.subtype}")

asyncio.run(main())

CortexCodeSDKClient

Pour les conversations interactives à plusieurs tours. Prend en charge le protocole de gestionnaire de contexte asynchrone.

from cortex_code_agent_sdk import CortexCodeSDKClient, CortexCodeAgentOptions, AssistantMessage

async with CortexCodeSDKClient(CortexCodeAgentOptions(permission_mode="bypassPermissions", allow_dangerously_skip_permissions=True)) as client:
    # First turn
    await client.query("What files are in this directory?")
    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 (context preserved)
    await client.query("Now refactor the main 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="")

Méthodes

Méthode

Description

connect()

Connectez-vous au processus CLI et gardez la session ouverte pour les tours ultérieurs.

query(prompt, session_id?)

Envoyer une invite à l’agent

receive_messages()

Traiter tous les messages de l’agent

receive_response()

Traiter les messages jusqu’à ResultMessage (inclus)

interrupt()

Envoyer un signal d’interruption

set_permission_mode(mode)

Modifier le mode d’autorisation pour les tours ultérieurs dans la conversation

set_model(model)

Modifier le modèle pendant la conversation

stop_task(task_id)

Arrêter une tâche de sous-agent en cours d’exécution

disconnect()

Se déconnecter du processus CLI

Options

Configuration transmise à query() ou CortexCodeSDKClient.

Option

Type

Par défaut

Description

cwd

str | Path | None

None

Répertoire de travail pour la session

model

str | None

None

Modèle à utiliser. Utilisez "auto" pour la sélection automatique, ou un identificateur spécifique comme "claude-sonnet-4-6". Voir Modèles pris en charge.

connection

str | None

None

Nom de la connexion Snowflake à partir des paramètres de la connexion CLI , généralement ~/.snowflake/connections.toml, avec ~/.snowflake/config.toml également pris en charge pour les configurations existantes. S’il est omis, la CLI utilise default_connection_name du fichier TOML. Voir Configuration des connexions.

profile

str | None

None

Nom du profil (chargements à partir de ~/.snowflake/cortex/profiles/)

permission_mode

PermissionMode | None

None

"default" | "autoAcceptPlans" | "plan" | "bypassPermissions". Remarque :"bypassPermissions" nécessite allow_dangerously_skip_permissions=True. En mode "plan",``AskUserQuestion`` et ExitPlanMode peuvent être acheminés par can_use_tool ; refuser ExitPlanMode garde la planification active et l’approuver fait quitter le mode de planification pour que les tours suivants reprennent les autorisations normales.

allow_dangerously_skip_permissions

bool

False

Indicateur de sécurité requis lors de l’utilisation de permission_mode="bypassPermissions". Cet indicateur seul ne contourne pas les autorisations ; il n’autorise le contournement que lorsqu’il est explicitement demandé via permission_mode.

allowed_tools

list[str]

[]

Outils pour approuver automatiquement sans invite

disallowed_tools

list[str]

[]

Outils pour toujours refuser

max_turns

int | None

None

Limiter le nombre de tours d’agents par requête

effort

str | None

None

Niveau d’effort du modèle : "minimal" | "low" | "medium" | "high" | "max"

system_prompt

str | SystemPromptPreset | None

None

Invite système personnalisée. Transmettez une chaîne pour un remplacement complet, ou {"type": "preset", "append": "extra text"} à ajouter à l’invite par défaut.

continue_conversation

bool

False

Poursuivre la session la plus récente au lieu d’en démarrer une nouvelle

resume

str | None

None

ID de session pour reprendre une conversation précédente.

fork_session

bool

False

Lors de la reprise, passer à un nouvel ID de session au lieu de poursuivre la session précédente

add_dirs

list[str | Path]

[]

Répertoires supplémentaires à ajouter au contexte de l’agent

env

dict[str, str]

{}

Variables d’environnement à transmettre au processus CLI

plugins

list[SdkPluginConfig]

[]

Configurations de plug-in. Prend actuellement en charge les plug-ins locaux : {"type": "local", "path": "/path/to/plugin"}

abort_event

asyncio.Event | None

None

Un :py:classe :asynchronio.Event qui, lorsqu’il est défini, envoie une interruption à l’agent en cours d’exécution. La session reste active pour d’autres invites. Équivalent Python de AbortController dans le SDK TypeScript.

mcp_servers

dict[str, McpServerConfig]

{}

Configurations de serveur MCP externes. Transmettez un dict qui mappe les noms de serveurs vers les configs stio, HTTP, ouSSE. Le transport SDK actuel prend en charge uniquement la configuration MCP basée sur dict.

hooks

dict | None

None

Gestionnaires d’événements de hook (voir Hooks)

can_use_tool

CanUseTool | None

None

Rappel de l’autorisation de l’outil personnalisé (voir Autorisations de l’outil)

include_partial_messages

bool

False

Inclure des événements de streaming au niveau des jetons ( StreamEvent)

output_format

dict | None

None

Format de sortie structuré. Exemple : {"type": "json_schema", "schema": {...}}

no_mcp

bool

False

Désactiver les serveurs MCP

session_id

str | None

None

ID de session explicite à utiliser

setting_sources

list[str] | None

None

Définition des sources à charger : "user" | "project" | "local"

cli_path

str | Path | None

os.environ.get("CORTEX_CODE_CLI_PATH") or "cortex"

Chemin vers le binaire CLI. S’il est omis, le SDK contrôle d’abord CORTEX_CODE_CLI_PATH et, sinon, revient à cortex sur PATH.

extra_args

dict[str, str | None]

{}

Indicateurs de CLI supplémentaires en tant que paires clé-valeur. Utilisez la valeur None pour les indicateurs booléens.

stderr

Callable[[str], None] | None

None

Rappel invoqué avec chaque ligne de sortie stderr CLI

Types de message

AssistantMessage

Émis lorsque l’agent produit une réponse. Contient un ou plusieurs blocs de contenu.

@dataclass
class AssistantMessage:
    content: list[ContentBlock]
    model: str
    parent_tool_use_id: str | None = None
    error: AssistantMessageError | None = None

AssistantMessageError est l’un ou l’autre des états suivants : "authentication_failed", "billing_error", "rate_limit", "invalid_request", "server_error", "unknown".

ResultMessage

Émis lorsque l’agent termine un tour. Contrôlez la réussite ou l’échec de subtype et is_error.

@dataclass
class ResultMessage:
    subtype: str
    duration_ms: int
    duration_api_ms: int
    is_error: bool
    num_turns: int
    session_id: str
    stop_reason: str | None = None
    total_cost_usd: float | None = None
    usage: dict[str, Any] | None = None
    result: str | None = None
    structured_output: Any = None

UserMessage

Renvoyé lorsqu’un message utilisateur est traité.

@dataclass
class UserMessage:
    content: str | list[ContentBlock]
    uuid: str | None = None
    parent_tool_use_id: str | None = None
    tool_use_result: dict[str, Any] | None = None

SystemMessage

Événements du système tels que l’initialisation des sessions et les mises à jour des tâches.

@dataclass
class SystemMessage:
    subtype: str
    data: dict[str, Any]

Le SDK fournit également des sous-classes spécialisées pour les messages système liés aux tâches :

Sous-classe

Description

TaskStartedMessage

Émis lors du démarrage d’une tâche de sous-agent. Champs : task_id, description, uuid, session_id, tool_use_id, task_type.

TaskProgressMessage

Émis pendant l’exécution d’une tâche. Champs : task_id, description, usage (TaskUsage), uuid, session_id, last_tool_name.

TaskNotificationMessage

Émis lorsqu’une tâche se termine, échoue ou est arrêtée. Champs : task_id, status ("completed" | "failed" | "stopped"), output_file, summary, uuid, session_id, usage.

Ces sous-classes étendent SystemMessage, afin que les contrôles isinstance(msg, SystemMessage) existants continuent de correspondre.

StreamEvent

Mises à jour partielles des messages pendant le streaming au niveau du jeton. Nécessite include_partial_messages=True.

@dataclass
class StreamEvent:
    uuid: str
    session_id: str
    event: dict[str, Any]   # Partial text/thinking stream event from Cortex Code
    parent_tool_use_id: str | None = None

StreamEvent est émis pour les blocs de texte partiels et les blocs de réflexion. Les appels complets des outils arrivent toujours comme blocs de contenu AssistantMessage et les résultats des outils arrivent toujours sous la forme de blocs de contenu UserMessage.

Blocs de contenu

Type

Champs

TextBlock

.type = "text", .text: str

ThinkingBlock

.type = "thinking", .thinking: str, .signature: str

ToolUseBlock

.type = "tool_use", .id: str, .name: str, .input: dict

ToolResultBlock

.type = "tool_result", .tool_use_id: str, .content: str | list | None, .is_error: bool | None

Hooks

Les hooks vous permettent d’intercepter les événements du cycle de vie à des fins de journalisation, de validation ou de comportement personnalisé.

Configuration des hooks

Les hooks sont configurés via l’option hooks sur CortexCodeAgentOptions. Chaque événement de hook correspond à une liste d’objets HookMatcher.

from cortex_code_agent_sdk import CortexCodeAgentOptions, HookMatcher

async def my_pre_tool_hook(input_data, tool_use_id, context):
    print(f"Tool {input_data['tool_name']} about to run")
    return {"continue_": True}

options = CortexCodeAgentOptions(
    hooks={
        "PreToolUse": [
            HookMatcher(
                matcher="Bash",          # Only match Bash tool, or None for all
                hooks=[my_pre_tool_hook],
                timeout=30.0,            # Timeout in seconds (default: 60)
            ),
        ],
    },
)

Signature de rappel du hook

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]
  • entrée : Entrée de hook à typage fort (voir tableau ci-dessous)

  • tool_use_id: Identificateur d’utilisation de l’outil facultatif

  • contexte : HookContext avec un champ signal (réservé pour la prise en charge future du signal d’annulation)

Événements de hooks

Événement

Input Type

Description

PreToolUse

PreToolUseHookInput

Avant l’exécution d’un outil. Champs : tool_name, tool_input, tool_use_id.

PostToolUse

PostToolUseHookInput

Une fois qu’un outil est terminé. Champs : tool_name, tool_input, tool_response, tool_use_id.

UserPromptSubmit

UserPromptSubmitHookInput

Lorsque l’utilisateur soumet une invite. Champs : prompt.

Stop

StopHookInput

Lorsque l’agent s’arrête. Champs : stop_hook_active.

SubagentStop

SubagentStopHookInput

Lorsqu’un sous-agent se termine. Champs : stop_hook_active.

Notification

NotificationHookInput

Sur les événements de notification. Champs : message, notification_type.

PermissionRequest

PermissionRequestHookInput

Lorsqu’un outil demande une autorisation. Champs : tool_name, tool_input.

PreCompact

PreCompactHookInput

Avant le compactage du contexte. Champs : trigger ("manual" | "auto"), custom_instructions.

Toutes les entrées de hook incluent des champs de base :session_id, transcript_path, cwd, et permission_mode en option.

Sortie du hook

Les rappels de hook renvoient un objet de sortie synchrone :

# Synchronous output
SyncHookJSONOutput = TypedDict("SyncHookJSONOutput", {
    "continue_": bool,               # Whether agent should proceed (default: True)
    "stopReason": str,                # Message shown when continue_ is False
    "decision": Literal["block"],     # Block the action
    "reason": str,                    # Feedback for the agent
    "hookSpecificOutput": ...,        # Event-specific controls
}, total=False)

Note

Le SDK Python utilise continue_ (avec un trait de soulignement de fin) pour éviter le conflit de mots-clés Python. La valeur est automatiquement convertie en continue lorsqu’il est envoyé à CLI.

Autorisations de l’outil

Le rappel can_use_tool vous permet de contrôler par programme les autorisations des outils.

Pour de nombreux contrôles d’autorisation d’outils ordinaires, l’entrée du rappel contient des champs tels que {"action": ..., "resource": ...}. Le résultat d’autorisation/de refus et le message de refus facultatif sont utilisés pour ces contrôles. updated_input est utilisé pour les pseudo-outils routés via le SDK tels que AskUserQuestion et ExitPlanMode, qui contiennent des champs spécifiques à l’outil.

from typing import Any

from cortex_code_agent_sdk import (
    CortexCodeAgentOptions,
    PermissionResultAllow,
    PermissionResultDeny,
    ToolPermissionContext,
)

async def my_permission_handler(
    tool_name: str,
    tool_input: dict[str, Any],
    context: ToolPermissionContext,
) -> PermissionResultAllow | PermissionResultDeny:
    if tool_name == "Write" and str(tool_input.get("resource", "")).endswith(".env"):
        return PermissionResultDeny(message="Writing env files is not allowed")
    return PermissionResultAllow()

options = CortexCodeAgentOptions(can_use_tool=my_permission_handler)

Types

CanUseTool = Callable[
    [str, dict[str, Any], ToolPermissionContext],
    Awaitable[PermissionResult],
]

@dataclass
class ToolPermissionContext:
    signal: Any | None = None
    blocked_path: str | None = None
    decision_reason: str | None = None
    tool_use_id: str = ""
    agent_id: str | None = None

@dataclass
class PermissionResultAllow:
    behavior: Literal["allow"] = "allow"
    updated_input: dict[str, Any] | None = None
    tool_use_id: str | None = None

@dataclass
class PermissionResultDeny:
    behavior: Literal["deny"] = "deny"
    message: str = ""
    interrupt: bool = False
    tool_use_id: str | None = None

Configuration du serveur MCP

L’option mcp_servers accepte un dict mappant les noms des serveurs vers les configurations de serveur MCP externes.

Type de configuration

Description

McpStdioServerConfig

Processus externes via stio. Champs : command, args, env.

McpSSEServerConfig

Événements envoyés par le serveur Champs : type="sse", url, headers.

McpHttpServerConfig

Transport HTTP. Champs : type="http", url, headers.

Exemple

options = CortexCodeAgentOptions(
    mcp_servers={
        "external": {
            "command": "npx",
            "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
         },
     },
 )

Traitement des erreurs

from cortex_code_agent_sdk import (
    query, CortexCodeAgentOptions, ResultMessage,
    CortexCodeSDKError, CLINotFoundError, CLIConnectionError,
)

try:
    async for message in query(prompt="...", options=CortexCodeAgentOptions()):
        if isinstance(message, ResultMessage) and message.is_error:
            print(f"Agent error ({message.subtype}): {message.result}")
except CLINotFoundError:
    print("Cortex CLI not found. Is it installed and on PATH?")
except CLIConnectionError:
    print("Failed to connect to CLI process")
except CortexCodeSDKError as e:
    print(f"SDK error: {e}")

Types d’erreur

Exception

Description

CortexCodeSDKError

Exception de base pour toutes les erreurs SDK

CLINotFoundError

Binaire CLI introuvable sur PATH

CLIConnectionError

Impossible de se connecter à ou de communiquer avec CLI

ProcessError

Le processus CLI s’est terminé de manière inattendue

CLIJSONDecodeError

Impossible d’analyser JSON depuis la sortie CLI