Cortex Code Agent SDK 참조 – Python

이 항목은 Python용 Cortex Code Agent SDK의 전체 API 참조를 제공합니다 — 모든 함수, 클래스, 유형을 포함합니다.

출처: Cortex Code Agent SDK reference – Python

본문

설치

pip install cortex-code-agent-sdk

Python 3.10 이상이 필요합니다. 의존성: anyio, mcp, typing_extensions. 패키지를 cortex_code_agent_sdk로 가져옵니다. SDK는 Cortex Code CLI가 별도로 설치되어 있다고 기대합니다. PATH에 없으면 CORTEX_CODE_CLI_PATH=/path/to/cortex를 설정하거나 CortexCodeAgentOptions에서 cli_path를 전달하세요.

함수

`query()`

Cortex Code와 상호작용하는 기본 함수. 도착하는 대로 메시지를 생성하는 async 이터레이터를 반환합니다.

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

매개변수

매개변수	유형	설명
`prompt`	str \| AsyncIterable[dict]	사용자 프롬프트 문자열, 또는 스트리밍 입력용 메시지 dict의 async iterable
`options`	CortexCodeAgentOptions \| None	구성 옵션. 기본값 `CortexCodeAgentOptions()`
`transport`	Transport \| None	사용자 정의 transport. 기본값 하위 프로세스 CLI transport

반환값

Message 객체(AssistantMessage, ResultMessage, UserMessage, SystemMessage, StreamEvent)를 생성하는 async 이터레이터.

예시

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`

다중 턴 대화형 대화용. async 컨텍스트 매니저 프로토콜을 지원합니다.

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

메서드

메서드	설명
`connect()`	CLI 프로세스에 연결하고 이후 턴을 위해 세션을 열어 둠
`query(prompt, session_id?)`	에이전트에 프롬프트 전송
`receive_messages()`	에이전트에서 모든 메시지 생성
`receive_response()`	`ResultMessage`까지(포함) 메시지 생성
`interrupt()`	인터럽트 신호 전송
`set_permission_mode(mode)`	대화의 이후 턴에 대한 권한 모드 변경
`set_model(model)`	대화 중 모델 변경
`stop_task(task_id)`	실행 중인 하위 에이전트 작업 중지
`disconnect()`	CLI 프로세스에서 연결 해제

옵션

query() 또는 CortexCodeSDKClient에 전달되는 구성.

옵션	유형	기본값	설명
`cwd`	str \| Path	None	세션의 작업 디렉터리
`model`	str	None	사용할 모델. 자동 선택에는 `"auto"`, 또는 `"claude-sonnet-4-6"` 같은 특정 식별자.
`connection`	str	None	Snowflake CLI 연결 설정의 Snowflake 연결 이름, 일반적으로 `~/.snowflake/connections.toml`, 기존 설정에 `~/.snowflake/config.toml`도 지원. 생략하면 CLI가 TOML 파일의 `default_connection_name`을 사용.
`profile`	str	None	프로필 이름(`~/.snowflake/cortex/profiles/`에서 로드)
`permission_mode`	PermissionMode	None	`"default"` \| `"autoAcceptPlans"` \| `"plan"` \| `"bypassPermissions"`. 참고: `"bypassPermissions"`는 `allow_dangerously_skip_permissions=True`가 필요. `"plan"` 모드에서 `AskUserQuestion`과 `ExitPlanMode`를 `can_use_tool`로 라우팅할 수 있고, `ExitPlanMode` 거부는 계획을 활성 상태로 유지하며 승인하면 플랜 모드를 종료해 이후 턴이 정상 권한을 재개합니다.
`allow_dangerously_skip_permissions`	bool	False	`permission_mode="bypassPermissions"` 사용 시 필요한 안전 플래그. 이 플래그만으로는 권한을 우회하지 않으며, permission_mode를 통해 명시적으로 요청할 때만 바이패스를 허용합니다.
`allowed_tools`	list[str]	[]	프롬프트 없이 자동 승인할 도구
`disallowed_tools`	list[str]	[]	항상 거부할 도구
`max_turns`	int	None	쿼리당 에이전트 턴 수 제한
`effort`	str	None	모델 노력 수준: `"minimal"` \| `"low"` \| `"medium"` \| `"high"` \| `"max"`
`system_prompt`	str \| SystemPromptPreset	None	사용자 정의 시스템 프롬프트. 완전한 재정의에는 문자열, 기본 프롬프트에 추가하려면 `{"type": "preset", "append": "extra text"}`.
`continue_conversation`	bool	False	새 세션을 시작하는 대신 가장 최근 세션 계속
`resume`	str	None	이전 대화를 재개할 세션 ID
`fork_session`	bool	False	재개 시 이전 세션을 계속하는 대신 새 세션 ID로 포크
`add_dirs`	list[str \| Path]	[]	에이전트 컨텍스트에 추가할 추가 디렉터리
`env`	dict[str, str]	{}	CLI 프로세스에 전달할 환경 변수
`plugins`	list[SdkPluginConfig]	[]	플러그인 구성. 현재 로컬 플러그인 지원: `{"type": "local", "path": "/path/to/plugin"}`
`abort_event`	asyncio.Event	None	설정되면 실행 중인 에이전트에 인터럽트를 보내는 `asyncio.Event`. 추가 프롬프트를 위해 세션은 계속 유지됩니다. TypeScript SDK의 `AbortController`의 Python 버전.
`mcp_servers`	dict[str, McpServerConfig]	{}	외부 MCP 서버 구성. 서버 이름을 stdio, HTTP 또는 SSE 구성에 매핑하는 dict를 전달. 현재 SDK transport는 dict 기반 MCP 구성만 지원.
`hooks`	dict	None	훅 이벤트 핸들러(Hooks 참고)
`can_use_tool`	CanUseTool	None	사용자 정의 도구 권한 콜백(Tool permissions 참고)
`include_partial_messages`	bool	False	토큰 수준 스트리밍 이벤트(`StreamEvent`) 포함
`output_format`	dict	None	구조화된 출력 형식. 예: `{"type": "json_schema", "schema": {...}}`
`no_mcp`	bool	False	MCP 서버 비활성화
`session_id`	str	None	사용할 명시적 세션 ID
`setting_sources`	list[str]	None	로드할 설정 소스: `"user"` \| `"project"` \| `"local"`
`cli_path`	str \| Path	`os.environ.get("CORTEX_CODE_CLI_PATH")` 또는 `"cortex"`	CLI 바이너리 경로. 생략하면 SDK가 먼저 `CORTEX_CODE_CLI_PATH`를 확인하고 아니면 PATH의 `cortex`로 폴백.
`extra_args`	dict[str, str \| None]	{}	키-값 쌍의 추가 CLI 플래그. 부울 플래그에는 None 값 사용.
`stderr`	Callable[[str], None]	None	CLI stderr 출력의 각 줄마다 호출되는 콜백

메시지 유형

`AssistantMessage`

에이전트가 응답을 만들 때 발생. 하나 이상의 콘텐츠 블록을 포함.

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

AssistantMessageError는 다음 중 하나: "authentication_failed", "billing_error", "rate_limit", "invalid_request", "server_error", "unknown".

`ResultMessage`

에이전트가 턴을 끝낼 때 발생. 성공/실패는 subtype과 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`

사용자 메시지가 처리될 때 다시 에코.

@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`

세션 초기화 및 작업 업데이트 같은 시스템 이벤트.

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

SDK는 작업 관련 시스템 메시지의 특화 하위 클래스도 제공합니다.

하위 클래스	설명
`TaskStartedMessage`	하위 에이전트 작업이 시작될 때 발생. 필드: `task_id`, `description`, `uuid`, `session_id`, `tool_use_id`, `task_type`.
`TaskProgressMessage`	작업이 실행되는 동안 발생. 필드: `task_id`, `description`, `usage`(TaskUsage), `uuid`, `session_id`, `last_tool_name`.
`TaskNotificationMessage`	작업이 완료, 실패 또는 중지될 때 발생. 필드: `task_id`, `status`(`"completed"` \| `"failed"` \| `"stopped"`), `output_file`, `summary`, `uuid`, `session_id`, `usage`.

이 하위 클래스는 SystemMessage를 확장하므로 기존 isinstance(msg, SystemMessage) 확인이 계속 일치합니다.

`StreamEvent`

토큰 수준 스트리밍 중 부분 메시지 업데이트. 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는 부분 텍스트 및 추론(thinking) 블록에 대해 발생합니다. 완전한 도구 호출은 여전히 AssistantMessage 콘텐츠 블록으로, 도구 결과는 여전히 UserMessage 콘텐츠 블록으로 도착합니다.

콘텐츠 블록

유형	필드
`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`

훅

훅을 사용하면 로깅, 검증 또는 사용자 정의 동작을 위해 수명 주기 이벤트를 가로챌 수 있습니다.

훅 구성

훅은 CortexCodeAgentOptions의 hooks 옵션으로 구성합니다. 각 훅 이벤트는 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)
            ),
        ],
    },
)

훅 콜백 시그니처

HookCallback = Callable[
    [HookInput, str | None, HookContext],
    Awaitable[HookJSONOutput],
]

input: 강력한 타입의 훅 입력(아래 표 참고)
tool_use_id: 선택적 도구 사용 식별자
context: signal 필드가 있는 HookContext(향후 abort 신호 지원용 예약)

훅 이벤트

이벤트	입력 유형	설명
`PreToolUse`	PreToolUseHookInput	도구 실행 전. 필드: `tool_name`, `tool_input`, `tool_use_id`.
`PostToolUse`	PostToolUseHookInput	도구 완료 후. 필드: `tool_name`, `tool_input`, `tool_response`, `tool_use_id`.
`UserPromptSubmit`	UserPromptSubmitHookInput	사용자가 프롬프트를 제출할 때. 필드: `prompt`.
`Stop`	StopHookInput	에이전트가 중지할 때. 필드: `stop_hook_active`.
`SubagentStop`	SubagentStopHookInput	하위 에이전트가 끝날 때. 필드: `stop_hook_active`.
`Notification`	NotificationHookInput	알림 이벤트에서. 필드: `message`, `notification_type`.
`PermissionRequest`	PermissionRequestHookInput	도구가 권한을 요청할 때. 필드: `tool_name`, `tool_input`.
`PreCompact`	PreCompactHookInput	컨텍스트 압축 전. 필드: `trigger`(`"manual"` \| `"auto"`), `custom_instructions`.

모든 훅 입력에는 기본 필드(session_id, transcript_path, cwd, 선택적으로 permission_mode)가 포함됩니다.

훅 출력

훅 콜백은 동기 출력 객체를 반환합니다.

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

참고: Python SDK는 Python 키워드 충돌을 피하기 위해 continue 대신 continue_(밑줄이 붙음)를 사용합니다. CLI에 보낼 때 자동으로 continue로 변환됩니다.

도구 권한

can_use_tool 콜백으로 도구 권한을 프로그래밍 방식으로 제어할 수 있습니다.

많은 일반 도구 권한 확인에서 콜백 입력에는 {"action": ..., "resource": ...} 같은 필드가 포함됩니다. 허용/거부 결과와 선택적 거부 메시지가 이러한 확인에 사용됩니다. updated_input은 AskUserQuestion, ExitPlanMode 같은 SDK 라우팅 의사 도구에 사용되며, 이들은 도구 특화 필드를 포함합니다.

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)

유형

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 서버 구성

mcp_servers 옵션은 서버 이름을 외부 MCP 서버 구성에 매핑하는 dict를 받습니다.

구성 유형	설명
`McpStdioServerConfig`	stdio를 통한 외부 프로세스. 필드: `command`, `args`, `env`.
`McpSSEServerConfig`	Server-sent events. 필드: `type="sse"`, `url`, `headers`.
`McpHttpServerConfig`	HTTP transport. 필드: `type="http"`, `url`, `headers`.

예시

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

오류 처리

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

오류 유형

예외	설명
`CortexCodeSDKError`	모든 SDK 오류의 기본 예외
`CLINotFoundError`	PATH에서 CLI 바이너리를 찾을 수 없음
`CLIConnectionError`	CLI에 연결하거나 통신 실패
`ProcessError`	CLI 프로세스가 예기치 않게 종료
`CLIJSONDecodeError`	CLI 출력에서 JSON 파싱 실패

Cortex Code Agent SDK 참조 – Python