collieai 1.1.0

Python SDK for CollieAi — safe customer-owned LLM streaming and input moderation.

CollieAi Python SDK

Safe customer-owned LLM streaming and pre-generation input moderation for CollieAi. You run your own model; the SDK checks the prompt before you call it and lets you stream back only CollieAi-released text — never raw model output.

Status: 1.1.0. Ships input moderation (with optional context analysis), streaming preflight, the ergonomic protect_stream(...) / protect_buffered(...) wrappers, the low-level streaming session, SSE delivery (session.stream_events + browser stream tokens), and provider adapters (collieai[openai] / collieai[anthropic]). Node/TypeScript (@collieai/sdk) and .NET (CollieAi.Client) SDKs mirror this one.

Install

pip install collieai

Quick start

Construct the client

from collieai import AsyncCollie

collie = AsyncCollie(
    api_key="clai_...",
    base_url="https://app.collieai.io",
    project_id="project_123",
)

The client reuses one pooled HTTP connection set. Close it when done (await collie.aclose()), or use it as an async context manager.

Check an input before calling your LLM

result = await collie.moderate.input(
    prompt=user_prompt,
    conversation_id=conversation_id,   # optional, groups a conversation
    correlation_id=chat_turn_id,       # optional, pins one turn
)

if result.blocked:
    return result.block_message or "Input blocked by policy."

A policy block is a normal result (result.blocked is True), not an exception. No webhook is required.

Analyze context alongside the prompt

Pass context to analyze the structured data (or raw string) you're about to feed the model — retrieved documents, tool output, a transaction record — alongside the prompt, so an injection hidden in that data is caught too, not just one in the prompt. Structured context travels as JSON; context_format ("auto" | "json" | "text") is an optional parsing hint for a raw string.

result = await collie.moderate.input(
    prompt=user_prompt,
    context={"transaction": {"title": retrieved_title, "memo": retrieved_memo}},
)

if result.context and result.context.status in ("monitored", "blocked", "degraded"):
    print(result.context.status,
          result.context.triggering_pointer,     # e.g. "/transaction/memo" — a path, never a value
          result.context.triggering_rule_type)

if result.blocked:
    # result.blocked_by is "prompt" or "context" — which surface blocked
    return result.block_message or "Blocked by policy."

result.context (a ContextModerationResult) carries the closed status enum (not_provided / disabled / not_run / clean / monitored / blocked / degraded), the triggering JSON Pointer + rule, and degraded-coverage markers (parse_degraded / limit_exceeded / inference_degraded). The pointer is a path, never a value — no analyzed content is echoed back.

The same context / context_format arguments work on protect_stream and protect_buffered when input checking is enabled (the default): context is analyzed by the input gate, so a wrapper called with check_input=False skips it entirely and never analyzes context. With the gate on, the verdict rides the terminal event (InputBlocked on a context block, Finished on success) and protect_buffered's result, and to_sse(...) serializes it onto the input_blocked / finished relay frames.

Until a policy turns context analysis on, context is inert — sending it is a safe no-op (status is not_provided / disabled). Enable it per policy server-side to activate detection.

Stream safely — the recommended path

protect_stream checks the input, calls your LLM only if it passes, batches the output, and yields only CollieAi-released events. Pass a factory (a zero-arg callable returning your stream), not an already-started stream — the SDK calls it once, after the input check.

from collieai import SafeDelta, Blocked, InputBlocked

async for event in collie.streaming.protect_stream(
    input=user_prompt,
    raw_stream_factory=lambda: your_llm_stream(user_prompt),
):
    if isinstance(event, SafeDelta):
        yield event.text                       # forward ONLY safe text
    elif isinstance(event, (Blocked, InputBlocked)):
        yield event.block_message or "Blocked by policy."
        break

A block — of the input or the output — is a normal terminal event, not an exception. You never touch chunk sequence numbers, retries, or batching.

FastAPI

from fastapi.responses import StreamingResponse
from openai import AsyncOpenAI
from collieai import SafeDelta, Blocked, InputBlocked
from collieai.adapters.openai import openai_factory   # pip install "collieai[openai]"

openai_client = AsyncOpenAI()

@app.post("/chat")
async def chat(prompt: str):
    factory = openai_factory(
        openai_client, model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
    )
    async def body():
        async for event in collie.streaming.protect_stream(
            input=prompt, raw_stream_factory=factory,
        ):
            if isinstance(event, SafeDelta):
                yield event.text
            elif isinstance(event, (Blocked, InputBlocked)):
                yield event.block_message or "Blocked by policy."
                return
    return StreamingResponse(body(), media_type="text/plain")

Choose the UX up front (preflight)

Ask whether the policy can stream before you call the LLM, and branch into a token-stream UI or a "checking response…" UI accordingly:

cap = await collie.streaming.preflight()   # cached until cap.valid_until

if cap.recommended_client_behavior == "stream":
    async for event in collie.streaming.protect_stream(
        input=user_prompt, raw_stream_factory=lambda: your_llm_stream(user_prompt),
    ):
        ...   # token-stream UI
elif cap.recommended_client_behavior == "buffer_then_show":
    result = await collie.streaming.protect_buffered(
        input=user_prompt, raw_stream_factory=lambda: your_llm_stream(user_prompt),
    )
    ...       # "checking response…" then show result
else:
    raise RuntimeError(cap.reason_detail or cap.reason)   # fail_fast

If you'd rather not branch yourself, pass require_streaming=True to protect_stream: it preflights first and raises BufferedFallbackRequired (policy must buffer) or a PreflightError (can't be served) before calling your LLM.

Buffered fallback

When a policy can't stream, check the whole response at once — same input-gate and factory contract, but it returns a single result instead of events:

result = await collie.streaming.protect_buffered(
    input=user_prompt,
    raw_stream_factory=lambda: your_llm_stream(user_prompt),
)
return result.block_message if result.blocked else result.filtered_text

Relay to a browser (SSE)

A backend that submits chunks for a job can also subscribe to that job's CollieAi SSE stream and relay safe events to a browser, with automatic reconnect:

from collieai import Blocked, Finished, StreamInterrupted, to_sse

async for event in session.stream_events():   # auto-resumes from Last-Event-ID
    if isinstance(event, StreamInterrupted):
        continue                              # reconnecting; nothing to forward
    yield to_sse(event)                       # SSE bytes for your text/event-stream
    if isinstance(event, (Blocked, Finished)):
        break

Replayed frames are deduplicated across reconnects, so a delta is never shown twice. Pass auto_resume=False to stop at the first StreamInterrupted and resume yourself with stream_events(last_event_id=...).

To let a browser subscribe directly, mint a short-lived, job-scoped token (your API key stays server-side):

st = await session.mint_stream_token()
# hand st.url to the browser: new EventSource(st.url)

The token is read-only, valid only for that one job's stream, and expires in st.expires_in seconds — re-mint before it lapses.

Low-level session (advanced)

If you need to drive batching yourself, use the session directly. It does not check the input — call moderate.input(...) first.

check = await collie.moderate.input(prompt=user_prompt)
if check.blocked:
    return check.block_message or "Input blocked by policy."

async with collie.streaming.session(input=user_prompt) as session:
    async for raw_delta in your_llm_stream():
        result = await session.push(raw_delta)
        for emit in result.emits:
            yield emit.text          # forward ONLY safe emits
        if result.finished:
            break
    await session.finish()

You never touch chunk sequence numbers, retries, or idempotency — the session owns them. Forward emit.text, never the raw model delta.

Errors

Catch typed exceptions instead of parsing strings. A few you'll see:

Exception	Meaning
ChunkRetryExhausted	transient failures exceeded the retry budget
ChunkPolicyChanged	policy changed mid-stream — start a new session
ChunkQuotaExceeded	rate-limited with no usable Retry-After
ChunkSessionFinished	the session already reached a terminal state
ConcurrentSessionUseError	overlapping push() calls on one session
ModerationError	moderate.input job failed/expired or timed out

All inherit from CollieError.

Retry behavior

The session retries the same chunk sequence on transient failures (network timeouts, 503, 504 chunk_filter_timeout, 429 with a usable Retry-After) with exponential backoff + jitter — default base 250 ms, max 4 s, 3 attempts per chunk, 10 s ceiling (AsyncCollie(retry_max_per_chunk_s=...)). A retried chunk never produces a duplicate visible emit.