Metadata-Version: 2.5
Name: snowflake-cortex-agent-sdk
Version: 0.0.3
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

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](https://pypi.org/project/snowflake-cortex-agent-sdk/).

```bash
pip install snowflake-cortex-agent-sdk
```

Requires Python 3.10 or newer.

## Quickstart

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

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

## Async

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": {"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.

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

| Resource | Purpose |
| --- | --- |
| `client.conversations` | Create, attach, list, and search conversations |
| `client.feedback` | Send feedback for a data-agent turn |

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

| Entry point | Purpose |
| --- | --- |
| `cortex_agent_sdk.cortexcode` | Python SDK for the local Cortex Code CLI: permissions, hooks, MCP, plugins, and session controls |

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

| Class | For | Header shape |
| --- | --- | --- |
| `PatAuth` (or string shorthand) | Programmatic Access Token | `Authorization: Bearer <token>` + `X-Snowflake-Authorization-Token-Type: PROGRAMMATIC_ACCESS_TOKEN` |
| `OAuthAuth` | External OAuth / Snowflake OAuth access tokens | `Authorization: Bearer <token>` + `X-Snowflake-Authorization-Token-Type: OAUTH` |
| `KeyPairJwtAuth` | Key-pair JWT auth | `Authorization: Bearer <token>` + `X-Snowflake-Authorization-Token-Type: KEYPAIR_JWT` |
| `SnowflakeSessionTokenAuth` | `snowflake-connector-python` session tokens | `Authorization: Snowflake Token="<token>"` |
| `CallableAuth` | Lazy token minting | Bearer + configurable token-type header |
| Custom `Authenticator` | Anything else | You return the header map |

## Status

Preview. Expect breaking changes as the underlying API stabilizes.

## License

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