snow-agent-sdk 0.1.0

Snow Agent Python SDK

Snow-Agent SDK

This README documents the same integration pattern used by consumers like genix-nova-be.

Consumer-style async usage (FastAPI)

Use SnowAgentClient in async routes/services and consume streaming events:

from snow_agent_sdk import ChunkEvent, DoneEvent, ErrorEvent, SnowAgentClient

SDK_TEST_SESSION_ID = "sdk-local-test-session"


async def run_local_snow_sdk_smoke_test(
    prompt: str,
    session_id: str,
) -> dict:
    client = SnowAgentClient(
        base_url="https://apps.solugenix.com/api/snow/service",
        token="<jwt-token-from-header-or-env>",
        session_id=session_id or SDK_TEST_SESSION_ID,
    )

    chunk_count = 0
    done_payload = None

    try:
        async for event in client.query(prompt):
            if isinstance(event, ChunkEvent):
                chunk_count += 1
            elif isinstance(event, DoneEvent):
                done_payload = {
                    "intent": event.intent,
                    "flow": event.flow,
                    "fallback": event.fallback,
                }
            elif isinstance(event, ErrorEvent):
                return {
                    "ok": False,
                    "error": event.message,
                    "chunk_count": chunk_count,
                }

        return {
            "ok": True,
            "chunk_count": chunk_count,
            "done": done_payload,
        }
    finally:
        await client.close()

How genix-nova-be uses it

• Calls SDK from api/ask.py as a non-blocking smoke test in /ask.
• Uses request session or header-provided session (x-snow-session-id) and passes it to SnowAgentClient(..., session_id=...).
• Iterates SSE events from client.query(prompt):
  • ChunkEvent -> counts/streams text chunks.
  • DoneEvent -> captures final metadata (intent, flow, fallback).
  • ErrorEvent -> returns/logs SDK error without crashing the API flow.
• Always closes the async client with await client.close() in finally.

Minimal sync usage

For scripts or blocking contexts, use SnowAgentClientSync:

from snow_agent_sdk import SnowAgentClientSync

client = SnowAgentClientSync(
    base_url="https://apps.solugenix.com/api/snow/service",
    token="<jwt-token-from-header-or-env>",
    session_id="optional-session-id",
)
text = client.ask("create incident for store 17 pos is offline")
print(text)

Notes

• Do not hardcode secrets/tokens in code.
• Keep base_url environment-specific (local, dev, qa, prod).
• Reuse session IDs per conversation when you want contextual continuity.