Snowflake Cortex Agent SDK for Python

The SDK centers on conversations, durable turns, and completed responses. It uses background runs internally while hiding run and thread bookkeeping.

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

client: CortexAgentClient = CortexAgentClient(account="my_account", auth="<token>")
conversation: Conversation = client.conversations.create(
    origin_application="cortex_agent_sdk",
    agent={"type": "inline", "spec": {"models": {"orchestration": "auto"}}},
)
turn: Turn = conversation.send("What is Snowflake Arctic?")
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.

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": {}}
    )
    turn: AsyncTurn = await conversation.send("Analyze this data.")
    answer: AgentResponse = await turn.result()

The root client exposes conversations and feedback. Advanced streaming and explicit run/thread management live under cortex_agent_sdk.api:

from cortex_agent_sdk.api import CortexAgentClient

api_client: CortexAgentClient = CortexAgentClient(
    account="my_account",
    auth="<token>",
)

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

Requires Python 3.10 or newer.