Metadata-Version: 2.5
Name: snowflake-cortex-agent-sdk
Version: 0.0.2
Summary: Python SDK for the Snowflake Cortex Agent REST API (agent:run streaming, threads, feedback)
Author-email: "Snowflake, Inc." <support@snowflake.com>
License: See LICENSE
License-File: LICENSE
Keywords: cortex,cortex-agent,openapi,sdk,snowflake
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.10
Requires-Dist: anyio>=4.0
Requires-Dist: httpx-sse>=0.4
Requires-Dist: httpx>=0.27
Requires-Dist: mcp<2,>=1.0
Requires-Dist: pydantic>=2.0
Requires-Dist: snowflake-connector-python[secure-local-storage]<5,>=4.7.1
Requires-Dist: typing-extensions>=4.0
Provides-Extra: dev
Requires-Dist: datamodel-code-generator>=0.28; extra == 'dev'
Requires-Dist: mypy>=1.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.20.0; extra == 'dev'
Requires-Dist: pytest-timeout>=2.0.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.6; extra == 'dev'
Description-Content-Type: text/markdown

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

```python
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:

```python
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:

```python
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`:

```python
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.
