Snow-Agent SDK

Python client for the snow-agent HTTP API (snow-service). It wraps the single natural-language endpoint POST /agent/query and its Server-Sent Events (SSE) contract: status, chunk, done (and error).

There is one interaction endpoint. The server classifies intent and runs the create / search / update flow; multi-turn confirmations (slot filling, duplicate yes/no, update confirmation) are just follow-up query() calls on the same session_id.

Package layout

Install from the repository root (local) or from PyPI package name snow-agent-sdk when published.

Authentication

snow-service uses bearer tokens. Pass the snt_live_… token you got from POST /tokens/register (self-service) or that an admin minted via POST /tokens:

Authorization: Bearer snt_live_…

The token is bound to a tenant server-side, so the working tenant is derived from it — you don't send working_tenant_id on normal calls. In code:

from snow_agent_sdk import BearerTokenAuth

auth = BearerTokenAuth(token=os.environ["SNOW_AGENT_TOKEN"])

agent_id is not carried by the token — agents are owned by the parent system. You supply it on the client (or per query() call).

Initialization flow

1. Choose base_url — e.g. https://<host>/api/snow/service (must match the deployment root_path + reverse proxy).
2. Token — pass token="snt_live_…" (or auth=BearerTokenAuth(...)).
3. agent_id — the globally-unique agent id from the parent system.
4. session_id — stable per conversation; required for multi-turn flows (slot filling, duplicate yes/no, update confirmation).
5. Construct — SnowAgentClient(base_url, token=..., agent_id=..., session_id=...).
6. Call query() — async iterator of StatusEvent, ChunkEvent, DoneEvent, or ErrorEvent.
7. Close — await client.close() in a finally block.

Methods (async SnowAgentClient)

Events (dataclasses):

• StatusEvent(stage=...) — progress (classifying_intent, checking_duplicates, …).
• ChunkEvent(text=...) — assistant text fragment.
• DoneEvent(intent=, flow=, fallback=) — terminal metadata.
• ErrorEvent(message=) — error surface.

Example: async

import os
from snow_agent_sdk import ChunkEvent, DoneEvent, ErrorEvent, SnowAgentClient


async def run(session_id: str) -> str:
    client = SnowAgentClient(
        base_url=os.environ["SNOW_AGENT_BASE_URL"].rstrip("/"),
        token=os.environ["SNOW_AGENT_TOKEN"],
        agent_id="agent_123",
        session_id=session_id,
        timeout=120.0,
    )
    parts: list[str] = []
    try:
        async for ev in client.query("create incident for store 44, printer offline"):
            if isinstance(ev, ChunkEvent):
                parts.append(ev.text)
            elif isinstance(ev, DoneEvent):
                break
            elif isinstance(ev, ErrorEvent):
                raise RuntimeError(ev.message)
        return "".join(parts)
    finally:
        await client.close()

Example: multi-turn confirmation (same session_id)

When the agent asks a question — a missing required field, "create anyway? (yes/no)" after finding similar tickets, or "apply this update? (yes/no)" — reply with another query() on the same session_id:

async for ev in client.query("yes", session_id=session_id):
    ...

The server holds the conversation state keyed by tenant:agent:session, so a bare "yes" / "no" resumes the pending flow — no need to resend the full incident text.

Sync client (SnowAgentClientSync)

Blocking wrapper that concatenates chunk text into a single string. Uses asyncio.run internally — do not call from inside a running event loop.

from snow_agent_sdk import SnowAgentClientSync

client = SnowAgentClientSync(
    base_url="https://apps.example.com/api/snow/service",
    token="snt_live_…",
    agent_id="agent_123",
    session_id="thread-123",
)
print(client.ask("search incidents for POS offline store 17"))

Error handling

• httpx.HTTPStatusError — not wrapped (e.g. 401 bad/missing token, 422 missing agent_id); enable logging around the stream.
• ErrorEvent — server sent an error SSE event; surface message.
• RuntimeError from SnowAgentClientSync — raised when an ErrorEvent is seen during collection.
• Timeouts — increase timeout= for long duplicate + rerank paths.

Extensibility — pass headers= (e.g. traceparent, X-Request-Id) to the constructor, or subclass SnowAgentClient.

Publishing / versioning

Bump the PyPI version in pyproject.toml when SDK behavior or types change.