snowflake-cortex-agent-sdk — Python SDK for Snowflake Cortex Agents

Version: v0.0.1 (preview)

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

• Lite + data agent runs (streaming, non-streaming, background)
• Named + versioned data agents (CRUD lifecycle)
• Coding-agent runs — sandbox, bash, skills, workspace mounts
• Threads (create / list / describe / update / delete / search) with auto-pagination
• Feedback
• Background run reconnect, cancel, and wait (client.runs.*)
• High-level Conversation abstraction
• Typed response views (AgentResponseView, ThreadMessageView)
• Sync and async clients as first-class twins
• Programmatic Cortex Code sessions through either the local CLI or Agent API

Request and response models are generated from the Cortex Agent API schema. The HTTP client, streaming runtime, resource classes, and higher-level helpers are hand-written for Python.

Install

pip install snowflake-cortex-agent-sdk

Requires Python 3.10+. Core dependencies include httpx, httpx-sse, and pydantic v2.

For Cortex Code sessions, install the code extra (including the optional Snowflake connector profile integration):

pip install 'snowflake-cortex-agent-sdk[code]'

Quickstart

import os
from cortex_agent_sdk import CortexAgentClient, EventType

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

for event in client.agent.run(
    {
        "models": {"orchestration": "claude-sonnet-4-5"},
        "messages": [{"role": "user", "content": [{"type": "text", "text": "Hi."}]}],
    }
):
    if event["event"] == EventType.RESPONSE_TEXT_DELTA:
        print(event.get("text", ""), end="", flush=True)

Async, same surface:

import asyncio
from cortex_agent_sdk import AsyncCortexAgentClient

async def main() -> None:
    async with AsyncCortexAgentClient(account="myaccount", auth=os.environ["SNOWFLAKE_PAT"]) as client:
        stream = client.agent.stream({"messages": [{"role": "user", "content": [{"type": "text", "text": "Hi."}]}]})
        async for text in stream.text_stream:
            print(text, end="", flush=True)

asyncio.run(main())

Or, for multi-turn without hand-managing ids:

from cortex_agent_sdk import Conversation

convo = Conversation.create(client, origin_application="my-app")
print(convo.ask("What is Snowflake Arctic?").text)
print(convo.ask("Name one advantage over Llama 3.").text)

Cortex Code sessions

The optional cortex_agent_sdk.cortexcode module provides one-shot queries and stateful Cortex Code sessions through either an installed Cortex Code CLI or the direct Agent API:

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

async def main() -> None:
    async for message in query(
        prompt="Review this repository for bugs",
        options=CortexCodeAgentOptions(connection="my_connection", mode="cli"),
    ):
        if isinstance(message, ResultMessage):
            print(message.result)

asyncio.run(main())

CLI mode supports local tools, hooks, permissions, MCP servers, and session resume or fork. API mode needs no local CLI and supports durable background runs, remote sandbox tools, cancellation, and thread-backed sessions.

Resources at a glance

Higher-level helpers:

• Conversation — hides thread_id + parent_message_id; ask() and ask_coding_agent() return AgentResponseView.
• BackgroundRun — the handle a background run returns; wait() and cancel(). Low-level live access stays on client.runs.resume().
• AgentResponseView / ThreadMessageView — typed content accessors.
• ThreadsPage — auto-paginating threads.list().

Auth at a glance

Every authenticator takes a callable as well as a string and is re-invoked per request attempt, so token rotation needs no extra plumbing.

Models and runtime data

Generated Pydantic v2 models are available from cortex_agent_sdk.models for construction-time validation and autocomplete. Request methods also accept plain dictionaries as an escape hatch for fields newer than the bundled schema.

Events, response content, and thread messages are returned as plain dictionaries so unknown server fields are preserved. Event validation is opt-in:

from cortex_agent_sdk.models import AgentRunRequest, parse_event

client.agent.run(AgentRunRequest(messages=[...], stream=False))   # validated
client.agent.run({"messages": [...], "stream": False})            # not validated

for event in client.agent.run({"messages": [...]}):
    typed = parse_event(event)      # model when known, dict when not

Status

Preview. Expect breaking changes while the underlying API stabilizes.

License

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