Referência do SDK do Cortex Code Agent – Python

Este tópico apresenta a referência completa da API para o SDK do Cortex Code Agent para Python, incluindo todas as funções, as classes e os tipos.

Instalação

pip install cortex-code-agent-sdk

Requer o Python 3.10 ou posterior. Dependências: anyio, mcp, typing_extensions. Importe o pacote como cortex_code_agent_sdk. O SDK espera que a Cortex Code CLI seja instalada separadamente. Se ela não estiver em seu PATH, defina CORTEX_CODE_CLI_PATH=/path/to/cortex ou passe cli_path em CortexCodeAgentOptions.

Funções

query()

A função primária para interagir com o Cortex Code. Retorna um iterador assíncrono que produz mensagens conforme elas chegam.

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

Parâmetros

Parâmetro

Tipo

Descrição

prompt

str | AsyncIterable[dict]

Cadeia de caracteres de prompt do usuário ou iterável assíncrono de dicionários de mensagem para entrada de streaming

options

CortexCodeAgentOptions | None

Opções de configuração. O padrão é CortexCodeAgentOptions()

transport

Transport | None

Transporte personalizado. O padrão é o transporte da CLI do subprocesso

Retornos

Um iterador assíncrono que produz objetos Message (AssistantMessage, ResultMessage, UserMessage, SystemMessage, StreamEvent).

Exemplo

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

Para conversas interativas multiturno. Compatível com o protocolo de gerenciador de contexto assíncrono.

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étodos

Método

Descrição

connect()

Conectar com o processo da CLI e manter a sessão aberta para turnos posteriores.

query(prompt, session_id?)

Enviar um prompt ao agente

receive_messages()

Produzir todas as mensagens do agente

receive_response()

Produzir mensagens até um ResultMessage (inclusive)

interrupt()

Enviar sinal de interrupção

set_permission_mode(mode)

Alterar o modo de permissão para turnos posteriores na conversa

set_model(model)

Alterar o modelo durante a conversa

stop_task(task_id)

Parar uma tarefa de subagente em execução

disconnect()

Desconectar do processo da CLI

Opções

Configuração passada para query() ou CortexCodeSDKClient.

Opção

Tipo

Padrão

Descrição

cwd

str | Path | None

None

Diretório de trabalho da sessão

model

str | None

None

Modelo a ser usado. Use "auto" para seleção automática ou um identificador específico, como "claude-sonnet-4-6". Consulte Modelos compatíveis.

connection

str | None

None

Nome da conexão do Snowflake conforme as configurações de conexão na Snowflake CLI, normalmente ~/.snowflake/connections.toml, com ~/.snowflake/config.toml também compatível com as configurações existentes. Se omitido, a CLI usa default_connection_name do arquivo TOML. Consulte Configurando as conexões.

profile

str | None

None

Nome do perfil (carrega de ~/.snowflake/cortex/profiles/)

permission_mode

PermissionMode | None

None

"default" | "autoAcceptPlans" | "plan" | "bypassPermissions". Observação: "bypassPermissions" requer allow_dangerously_skip_permissions=True. No modo "plan", é possível rotear AskUserQuestion e ExitPlanMode por can_use_tool. Negar ExitPlanMode mantém o planejamento ativo e aprová-lo sai do modo de plano para que os turnos posteriores retomem as permissões normais.

allow_dangerously_skip_permissions

bool

False

Sinalizador de segurança necessário ao usar permission_mode="bypassPermissions". Esse sinalizador sozinho não ignora as permissões; ele apenas permite o bypass quando solicitado explicitamente por permission_mode.

allowed_tools

list[str]

[]

Ferramentas para aprovação automática sem prompt

disallowed_tools

list[str]

[]

Ferramentas para negar sempre

max_turns

int | None

None

Limitar o número de turnos de agente por consulta

effort

str | None

None

Nível de esforço do modelo: "minimal" | "low" | "medium" | "high" | "max"

system_prompt

str | SystemPromptPreset | None

None

Prompt do sistema personalizado. Passe uma cadeia de caracteres para a substituição completa ou {"type": "preset", "append": "extra text"} para anexar ao prompt padrão.

continue_conversation

bool

False

Continuar a sessão mais recente em vez de iniciar uma nova

resume

str | None

None

ID da sessão para retomar uma conversa anterior

fork_session

bool

False

Ao retomar, bifurcar para um ID de nova sessão em vez de continuar a sessão anterior

add_dirs

list[str | Path]

[]

Diretórios adicionais para adicionar ao contexto do agente

env

dict[str, str]

{}

Variáveis de ambiente a serem passadas ao processo da CLI

plugins

list[SdkPluginConfig]

[]

Configurações de plug-in. Compatível no momento com plug-ins locais: {"type": "local", "path": "/path/to/plugin"}

abort_event

asyncio.Event | None

None

Um asyncio.Event que, quando definido, envia uma interrupção ao agente em execução. A sessão continua ativa para receber outros prompts. Equivalente em Python do AbortController no SDK TypeScript.

mcp_servers

dict[str, McpServerConfig]

{}

Configurações do servidor MCP externo. Passe um dict que mapeie nomes de servidores para configurações stdio, HTTP ou SSE. O transporte do SDK atual é compatível apenas com a configuração de MCP baseada em dict.

hooks

dict | None

None

Manipuladores de eventos de gancho (consulte Ganchos)

can_use_tool

CanUseTool | None

None

Retorno de chamada de permissão de ferramenta personalizada (consulte Permissões de ferramenta)

include_partial_messages

bool

False

Incluir eventos de streaming no nível do token (StreamEvent)

output_format

dict | None

None

Formato de saída estruturado. Exemplo: {"type": "json_schema", "schema": {...}}

no_mcp

bool

False

Desabilitar servidores MCP

session_id

str | None

None

ID de sessão explícito a ser usado

setting_sources

list[str] | None

None

Fontes de configuração a serem carregadas: "user" | "project" | "local"

cli_path

str | Path | None

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

Caminho para o binário da CLI. Se omitido, o SDK primeiro verifica``CORTEX_CODE_CLI_PATH`` e, caso contrário, retorna para cortex em PATH.

extra_args

dict[str, str | None]

{}

Sinalizadores de CLI adicionais como pares chave-valor. Use o valor None para sinalizadores boolianos.

stderr

Callable[[str], None] | None

None

Retorno de chamada invocado com cada linha da saída stderr da CLI

Tipos de mensagem

AssistantMessage

Emitida quando o agente produz uma resposta. Contém um ou mais blocos de conteúdo.

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

AssistantMessageError é uma destas opções: "authentication_failed", "billing_error", "rate_limit", "invalid_request", "server_error", "unknown".

ResultMessage

Emitida quando o agente termina um turno. Verifique subtype e is_error para sucesso ou falha.

@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

Repetida quando uma mensagem de usuário é processada.

@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

Eventos do sistema, como inicialização de sessão e atualizações de tarefas.

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

O SDK também fornece subclasses especializadas para mensagens do sistema relacionadas a tarefas:

Subclasse

Descrição

TaskStartedMessage

Emitida quando uma tarefa de subagente é iniciada. Campos: task_id, description, uuid, session_id, tool_use_id, task_type.

TaskProgressMessage

Emitida enquanto uma tarefa está em execução. Campos: task_id, description, usage (TaskUsage), uuid, session_id, last_tool_name.

TaskNotificationMessage

Emitida quando uma tarefa é concluída, falha ou é interrompida. Campos: task_id, status ("completed" | "failed" | "stopped"), output_file, summary, uuid, session_id, usage.

Essas subclasses estendem SystemMessage, portanto, as verificações isinstance(msg, SystemMessage) existentes de continuam correspondendo.

StreamEvent

Atualizações parciais de mensagens durante o streaming no nível do token. Requer 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 é emitido para texto parcial e blocos de raciocínio. Chamadas de ferramentas completas ainda chegam como blocos de conteúdo AssistantMessage, e resultados de ferramentas ainda chegam como blocos de conteúdo UserMessage.

Blocos de conteúdo

Tipo

Campos

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

Ganchos

Os ganchos permitem interceptar eventos de ciclo de vida para registro em log, validação ou comportamento personalizado.

Configurando ganchos

Os ganchos são configurados por meio da opção hooks em CortexCodeAgentOptions. Cada evento de gancho é mapeado para uma lista de objetos 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)
            ),
        ],
    },
)

Assinatura de retorno de chamada de gancho

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]
  • input: entrada de gancho fortemente tipada (consulte a tabela abaixo)

  • tool_use_id: identificador opcional de uso da ferramenta

  • context: HookContext com um campo signal (reservado para suporte futuro ao sinal de anulação)

Eventos de gancho

Evento

Tipo de entrada

Descrição

PreToolUse

PreToolUseHookInput

Antes que uma ferramenta seja executada. Campos: tool_name, tool_input, tool_use_id.

PostToolUse

PostToolUseHookInput

Depois que uma ferramenta é concluída. Campos: tool_name, tool_input, tool_response, tool_use_id.

UserPromptSubmit

UserPromptSubmitHookInput

Quando o usuário envia um prompt. Campos: prompt.

Stop

StopHookInput

Quando o agente para. Campos: stop_hook_active.

SubagentStop

SubagentStopHookInput

Quando um subagente é concluído. Campos: stop_hook_active.

Notification

NotificationHookInput

Em eventos de notificação. Campos: message, notification_type.

PermissionRequest

PermissionRequestHookInput

Quando uma ferramenta solicita permissão. Campos: tool_name, tool_input.

PreCompact

PreCompactHookInput

Antes da compactação do contexto. Campos: trigger ("manual" | "auto"), custom_instructions.

Todas as entradas de gancho incluem campos base: session_id, transcript_path, cwd e, opcionalmente, permission_mode.

Saída do gancho

Os retornos de chamada de gancho geram um objeto de saída síncrona:

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

Nota

O Python SDK usa continue_ (com um sublinhado final) para evitar conflito de palavras-chave do Python. Isso é convertido automaticamente em continue quando enviado para a CLI.

Permissões de ferramenta

O retorno de chamada can_use_tool permite que você controle programaticamente as permissões da ferramenta.

Para muitas verificações comuns de permissão de ferramentas, a entrada do retorno de chamada contém campos como {"action": ..., "resource": ...}. O resultado de permitir/negar e a mensagem de negação opcional são usados nestas verificações. updated_input é usado para pseudoferramentas roteadas por SDK, como AskUserQuestion e ExitPlanMode, que contêm campos específicos da ferramenta.

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)

Tipos

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

Configuração do servidor MCP

A opção mcp_servers aceita um dict que mapeia nomes de servidores para configurações de servidor MCP externo:

Tipo de configuração

Descrição

McpStdioServerConfig

Processo externo via stdio. Campos: command, args, env.

McpSSEServerConfig

Eventos enviados pelo servidor. Campos: type="sse", url, headers.

McpHttpServerConfig

Transporte HTTP. Campos: type="http", url, headers.

Exemplo

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

Tratamento de erros

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

Tipos de erro

Exceção

Descrição

CortexCodeSDKError

Exceção básica para todos os erros de SDK

CLINotFoundError

Binário da CLI não encontrado em PATH

CLIConnectionError

Falha na conexão ou comunicação com a CLI

ProcessError

O processo da CLI foi encerrado inesperadamente

CLIJSONDecodeError

Falha ao analisar JSON da saída da CLI