Snowflake Cortex Agent SDK for Python

Python SDK for the Snowflake Cortex Agent REST API. Covers the current public surface:

• Durable lite, data-agent, and coding-agent conversations
• Named + versioned data agents through the direct API client
• Coding-agent runs with sandbox, bash, skills, and workspace mounts
• Threads, background runs, streaming, and feedback through cortex_agent_sdk.api
• High-level Conversation.send() -> Turn abstraction
• Route-test helpers in cortex_agent_sdk.testing (see docs/13-testing.md)
• Curated AgentResponse content and usage view
• Synchronous and asyncio clients
• Claude-shaped local Cortex Code sessions via cortex_agent_sdk.cortexcode

Install

View snowflake-cortex-agent-sdk on PyPI.

pip install snowflake-cortex-agent-sdk

Requires Python 3.10 or newer.

Quickstart

import os
from cortex_agent_sdk import (
    AgentResponse,
    Conversation,
    ConversationMessage,
    CortexAgentClient,
    Turn,
)

client: CortexAgentClient = CortexAgentClient(
    account="my_account",
    auth=os.environ["SNOWFLAKE_PAT"],
)

conversation: Conversation = client.conversations.create(
    origin_application="cortex_agent_sdk",
    agent={"type": "inline", "spec": {"models": {"orchestration": "claude-sonnet-4-5"}}},
)

turn: Turn = conversation.send("What is Snowflake Arctic?")
print(f"run: {turn.id}")

answer: AgentResponse = turn.result()
print(answer.text)

history: list[ConversationMessage] = conversation.history()

Set a default agent when creating a conversation. A send can override it for one turn when needed:

inline_turn: Turn = conversation.send(
    "Summarize the context.",
    agent={"type": "inline", "spec": {"models": {"orchestration": "auto"}}},
)

data_turn: Turn = conversation.send(
    "Which products grew fastest?",
    agent={
        "type": "data",
        "spec": {"database": "ANALYTICS", "schema": "PUBLIC", "name": "SALES_AGENT"},
    },
)

coding_turn: Turn = conversation.send(
    "Fix the failing test.",
    agent={
        "type": "coding",
        "spec": {
            "models": {"orchestration": "claude-sonnet-4-5"},
            "permission_policy": "always_allow",
            "workspace_mounts": [{"name": "USER$MY_USER.PUBLIC.DEFAULT$"}],
        },
    },
)

Use turn.events() for event-level progress and turn.cancel() to stop a run.

Async

The asynchronous API mirrors the sync API:

from cortex_agent_sdk import AgentResponse, AsyncConversation, AsyncCortexAgentClient, AsyncTurn

async with AsyncCortexAgentClient(account="my_account", auth="<token>") as client:
    conversation: AsyncConversation = await client.conversations.create(
        origin_application="cortex_agent_sdk",
        agent={"type": "inline", "spec": {"models": {"orchestration": "auto"}}},
    )
    turn: AsyncTurn = await conversation.send("Analyze this data.")
    answer: AgentResponse = await turn.result()

Cortex Code sessions (cortex_agent_sdk.cortexcode)

cortex_agent_sdk.cortexcode is the Cortex Code SDK for driving the local cortex binary from Python. It exposes one-shot query() calls and long-lived CortexCodeSDKClient sessions with permissions, hooks, MCP, plugins, structured output, and session controls.

import asyncio
from cortex_agent_sdk.cortexcode import AssistantMessage, ResultMessage, query

async def main() -> None:
    async for message in query(prompt="Summarize this repository"):
        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())

This entry point runs the local Cortex Code CLI. For server-side integrations, use the package-root CortexAgentClient with a coding-agent conversation instead.

Resources at a glance

Conversation binds a thread to an inline, data, or coding agent and hides message and run plumbing. send() returns a Turn with id, result(), events(), and cancel().

For direct REST request shapes, streaming modes, raw response wrappers, or SSE parsing, import CortexAgentClient and related types from cortex_agent_sdk.api.

Separate entry point, not a client.* resource:

Auth at a glance

Authentication headers are rebuilt for every attempt. refresh_auth runs only for recognized token/session expiry responses, including legacy 200 JSON error envelopes; by default the SDK refreshes once and replays the request once. SDK-generated errors do not include credential values.

Status

Preview. Expect breaking changes as the underlying API stabilizes.

License

See LICENSE. Use is governed by your Snowflake customer agreement.