Cortex Code Agent SDK-Referenz – Python

Unter diesem Thema finden Sie die vollständige API-Referenz für das Cortex Code Agent SDK für Python, einschließlich aller Funktionen, Klassen und Typen.

Installation

pip install cortex-code-agent-sdk

Erfordert Python 3.10 oder höher. Abhängigkeiten: anyio, mcp, typing_extensions. Importieren Sie das Paket als cortex_code_agent_sdk. Das SDK erwartet, dass die Cortex Code CLI separat installiert wird. Wenn diese sich nicht in Ihrem PATH befindet, legen Sie CORTEX_CODE_CLI_PATH=/path/to/cortex fest oder übergeben Sie den cli_path in CortexCodeAgentOptions.

Funktionen

query()

Die Primärfunktion für die Interaktion mit Cortex Code. Gibt einen asynchronen Iterator zurück, der Meldungen sofort ausgibt, sobald sie eintreffen.

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

Parameter

Parameter

Typ

Beschreibung

prompt

str | AsyncIterable[dict]

Benutzer-Eingabeaufforderungszeichenfolge oder asynchrones Iterations-Objekt aus Message-Dicts für die Streaming-Eingabe

options

CortexCodeAgentOptions | None

Konfigurationsoptionen. Die Standardeinstellung ist CortexCodeAgentOptions().

transport

Transport | None

Kundenspezifischer Transport. Standardmäßig wird Subprozess-CLI-Transport verwendet

Rückgabewerte

Ein asynchroner Iterator, der Message-Objekte ausgibt (AssistantMessage, ResultMessage, UserMessage, SystemMessage, StreamEvent).

Beispiel

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

Für interaktive Multi-Turn-Konversationen. Unterstützt das asynchrone Kontextmanagerprotokoll.

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

Methoden

Methode

Beschreibung

connect()

Stellt eine Verbindung zum CLI-Prozess her und hält die Sitzung für später offen

query(prompt, session_id?)

Sendet eine Eingabeaufforderung an den Agenten

receive_messages()

Gibt alle Meldungen des Agenten aus

receive_response()

Ausgabe von Meldungen bis zu einem ResultMessage (einschließlich)

interrupt()

Sendet ein Interrupt-Signal

set_permission_mode(mode)

Ändert den Berechtigungsmodus für spätere Konversations-Turns

set_model(model)

Ändert das Modell während der Konversation

stop_task(task_id)

Stoppt eine laufende Subagentenaufgabe

disconnect()

Trennt die Verbindung zum CLI-Prozess

Optionen

Konfiguration, die an query() oder CortexCodeSDKClient übergeben wird.

Option

Typ

Standard

Beschreibung

cwd

str | Path | None

None

Arbeitsverzeichnis für die Sitzung

model

str | None

None

Zu verwendendes Modell. Verwenden Sie "auto" für die automatische Auswahl oder einen bestimmten Bezeichner wie "claude-sonnet-4-6". Siehe Unterstützte Modelle.

connection

str | None

None

Snowflake-Verbindungsname von Snowflake CLI-Verbindungseinstellungen, normalerweise ~/.snowflake/connections.toml; wird mit ~/.snowflake/config.toml auch für bestehende Setups unterstützt. Beim Weglassen verwendet die CLI den default_connection_name aus der TOML-Datei. Siehe Konfigurieren von Verbindungen.

profile

str | None

None

Profilname (wird aus ~/.snowflake/cortex/profiles/ geladen)

permission_mode

PermissionMode | None

None

"default" | "autoAcceptPlans" | "plan" | "bypassPermissions". Hinweis: "bypassPermissions" erfordert allow_dangerously_skip_permissions=True. Im "plan"-Modus können AskUserQuestion und ExitPlanMode über can_use_tool geroutet werden. Ablehnen von ExitPlanMode hält die Planung aktiv, und die Genehmigung beendet den Planmodus, sodass spätere Turns die normalen Berechtigungen wieder aufnehmen.

allow_dangerously_skip_permissions

bool

False

Sicherheitsflag, erforderlich bei Verwendung von permission_mode="bypassPermissions". Dieses Flag allein umgeht die Berechtigungen nicht. Es erlaubt die Umgehung nur, wenn dies explizit über permission_mode angefordert wird.

allowed_tools

list[str]

[]

Tools zur automatischen Genehmigung ohne Eingabeaufforderung

disallowed_tools

list[str]

[]

Tools, um immer abzulehnen

max_turns

int | None

None

Begrenzt die Anzahl der Agent-Turns pro Abfrage

effort

str | None

None

Effort-Level (Aufwandsstufe) des Modells: "minimal" | "low" | "medium" | "high" | "max"

system_prompt

str | SystemPromptPreset | None

None

Kundenspezifische System-Eingabeaufforderung. Übergeben Sie eine Zeichenfolge für eine vollständige Überschreibung oder {"type": "preset", "append": "extra text"} zum Anhängen an die Standardeingabeaufforderung.

continue_conversation

bool

False

Setzt die letzte Sitzung fort, anstatt eine neue zu beginnen

resume

str | None

None

Sitzungs-ID, um eine frühere Konversation fortzusetzen

fork_session

bool

False

Wechselt beim Fortsetzen zu einer neuen Sitzungs-ID, anstatt die vorherige Sitzung fortzusetzen

add_dirs

list[str | Path]

[]

Zusätzliche Verzeichnisse, die dem Kontext des Agenten hinzugefügt werden sollen

env

dict[str, str]

{}

Umgebungsvariablen, die an den CLI-Prozess übergeben werden

plugins

list[SdkPluginConfig]

[]

Plugin-Konfigurationen. Unterstützt derzeit lokale Plugins: {"type": "local", "path": "/path/to/plugin"}

abort_event

asyncio.Event | None

None

Ein :py:klasse:`asyncio.Event`, das, wenn es gesetzt ist, eine Unterbrechung an den ausgeführten Agenten sendet. Die Sitzung bleibt für weitere Eingabeaufforderungen bestehen. Python-Äquivalent von AbortController im TypeScript-SDK.

mcp_servers

dict[str, McpServerConfig]

{}

Externe MCP-Serverkonfigurationen. Übergeben Sie ein dict, das Servernamen zu stdio, HTTP- oderSSE-Konfigurationen zuordnet. Der aktuelle SDK-Transport unterstützt nur dict-basierte MCP-Konfiguration.

hooks

dict | None

None

Hook-Ereignis-Handler (siehe Hooks)

can_use_tool

CanUseTool | None

None

Kundenspezifischer Berechtigungs-Callback für Tools (siehe Tool-Berechtigungen)

include_partial_messages

bool

False

Schließt Streaming-Ereignisse auf Token-Ebene ein (StreamEvent)

output_format

dict | None

None

Strukturiertes Ausgabeformat. Beispiel: {"type": "json_schema", "schema": {...}}

no_mcp

bool

False

Deaktiviert MCP-Server

session_id

str | None

None

Explizite Sitzungs-ID verwenden

setting_sources

list[str] | None

None

Einstellen der zu ladenden Quellen: "user" | "project" | "local"

cli_path

str | Path | None

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

Pfad zu CLI-Binärdateien. Beim Weglassen überprüft das SDK zuerst CORTEX_CODE_CLI_PATH und greift ansonsten auf cortex im PATH zurück.

extra_args

dict[str, str | None]

{}

Zusätzliche CLI-Flags als Schlüssel-Wert-Paare. Verwenden Sie den None-Wert für boolesche Flags.

stderr

Callable[[str], None] | None

None

Callback, der mit jeder Zeile der CLI-Ausgabe „stderr“ aufgerufen wird.

Meldungstyp

AssistantMessage

Wird ausgegeben, wenn der Agent eine Antwort erzeugt. Enthält einen oder mehrere Inhaltsblöcke.

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

AssistantMessageError ist einer der folgenden Werte: "authentication_failed", "billing_error", "rate_limit", "invalid_request", "server_error", "unknown".

ResultMessage

Wird ausgegeben, wenn der Agent einen Turn beendet hat. Überprüfen Sie subtype und is_error auf Erfolg oder Misserfolg.

@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

Wird als Echo zurückgegeben, wenn eine Benutzermeldung verarbeitet wird.

@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

Systemereignisse wie Sitzungsinitialisierung und Aufgabenaktualisierungen.

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

Das SDK bietet auch spezielle Unterklassen für aufgabenbezogene Systemmeldungen:

Unterklasse

Beschreibung

TaskStartedMessage

Wird ausgegeben, wenn eine Subagentenaufgabe gestartet wird. Felder: task_id, description, uuid, session_id, tool_use_id, task_type.

TaskProgressMessage

Wird ausgegeben, während eine Aufgabe ausgeführt wird. Felder: task_id, description, usage (TaskUsage), uuid, session_id, last_tool_name.

TaskNotificationMessage

Wird ausgegeben, wenn eine Aufgabe abgeschlossen ist, fehlschlägt oder gestoppt wird. Felder: task_id, status ("completed" | "failed" | "stopped"), output_file, summary, uuid, session_id, usage.

Diese Unterklassen erweitern SystemMessage, sodass vorhandene isinstance(msg, SystemMessage)-Überprüfungen weiterhin übereinstimmen.

StreamEvent

Teilweise Meldungsaktualisierung beim Streaming auf Token-Ebene. Erfordert 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 wird für Teiltext- und Denkblöcke ausgegeben. Vollständige Toolaufrufe gehen weiterhin als AssistantMessage-Inhaltsblöcke ein, und die Ergebnisse des Tools kommen immer noch als UserMessage-Inhaltsblöcke an.

Inhaltsblöcke

Typ

Felder

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

Mit Hooks können Sie Lebenszyklusereignisse für die Protokollierung, Validierung oder kundenspezifisches Verhalten abfangen.

Konfigurieren von -Hooks

Hooks werden über die Option hooks unter CortexCodeAgentOptions konfiguriert. Jedes Hook-Ereignis wird einer Liste von HookMatcher-Objekten zugeordnet.

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)
            ),
        ],
    },
)

Signatur des Hook-Callbacks

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]
  • Eingabe: Stark typisierte Hook-Eingabe (siehe Tabelle unten)

  • tool_use_id: Optionaler Bezeichner für die Toolverwendung

  • Kontext: HookContext mit einem signal-Feld (reserviert für zukünftige Unterstützung von Abbruchsignaturen)

Hook-Ereignisse

Ereignis

Input Type

Beschreibung

PreToolUse

PreToolUseHookInput

Bevor ein Tool ausgeführt wird. Felder: tool_name, tool_input, tool_use_id.

PostToolUse

PostToolUseHookInput

Nachdem ein Tool abgeschlossen ist. Felder: tool_name, tool_input, tool_response, tool_use_id.

UserPromptSubmit

UserPromptSubmitHookInput

Wenn der Benutzer eine Eingabeaufforderung übermittelt. Felder: prompt.

Stop

StopHookInput

Wenn der Agent stoppt. Felder: stop_hook_active.

SubagentStop

SubagentStopHookInput

Wenn ein Subagent beendet ist. Felder: stop_hook_active.

Notification

NotificationHookInput

Bei Benachrichtigungsereignissen. Felder: message, notification_type.

PermissionRequest

PermissionRequestHookInput

Wenn ein Tool eine Berechtigung anfragt. Felder: tool_name, tool_input.

PreCompact

PreCompactHookInput

Vor der Kontextkomprimierung. Felder: trigger ("manual" | "auto"), custom_instructions.

Alle Hook-Eingaben enthalten Basisfelder: session_id, transcript_path, cwd und optional permission_mode.

Hook-Ausgabe

Hook-Callbacks geben ein synchrones Ausgabeobjekt zurück:

# 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)

Bemerkung

Das Python-SDK verwendet continue_ (mit einem nachstehenden Unterstrich), um einen Konflikt mit dem Python-Schlüsselwort zu vermeiden. Dies wird beim Senden an die CLI automatisch in continue umgewandelt.

Tool-Berechtigungen

Mit dem can_use_tool-Callback können Sie die Berechtigungen von Tools programmgesteuert steuern.

Bei vielen normalen Tool-Berechtigungsprüfungen enthält die Callback-Eingabe Felder wie {"action": ..., "resource": ...}. Für diese Prüfungen werden das Zulassen/Ablehnen-Ergebnis und die optionale Ablehnungsmeldung verwendet. updated_input wird für SDK-Routing-Pseudo-Tools wie AskUserQuestion und``ExitPlanMode`` verwendet, die Tool-spezifische Felder enthalten.

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)

Typen

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

MCP-Serverkonfiguration

Die Option mcp_servers akzeptiert ein „dict“, das Servernamen externen MCP-Serverkonfigurationen zuordnet:

Konfigurationstyp

Beschreibung

McpStdioServerConfig

Externer Prozess über stdio. Felder: command, args, env.

McpSSEServerConfig

Vom Server gesendete Ereignisse. Felder: type="sse", url, headers.

McpHttpServerConfig

HTTP-Transport. Felder: type="http", url, headers.

Beispiel

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

Fehlerbehandlung

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

Fehlertypen

Ausnahme

Beschreibung

CortexCodeSDKError

Basisausnahme für alle SDK-Fehler

CLINotFoundError

CLI-Binärdatei nicht in PATH gefunden

CLIConnectionError

Eine Verbindung zu oder Kommunikation mit CLI ist fehlgeschlagen

ProcessError

CLI-Prozess wurde unerwartet beendet

CLIJSONDecodeError

Parsen von JSON aus CLI-Ausgabe ist fehlgeschlagen