Metadata-Version: 2.4
Name: collieai
Version: 0.1.0
Summary: Python SDK for CollieAi — safe customer-owned LLM streaming and input moderation.
Author: CollieAi
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: collieai,guardrails,llm,moderation,streaming
Requires-Python: >=3.9
Requires-Dist: httpx<1,>=0.27
Requires-Dist: pydantic<3,>=2
Provides-Extra: anthropic
Requires-Dist: anthropic<1,>=0.39; extra == 'anthropic'
Provides-Extra: dev
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8; extra == 'dev'
Provides-Extra: openai
Requires-Dist: openai<2,>=1; extra == 'openai'
Description-Content-Type: text/markdown

# CollieAi Python SDK

Safe customer-owned LLM streaming and pre-generation input moderation for
[CollieAi](https://app.collieai.io). 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: **0.1.0.** Ships input moderation, 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]`).
> A Node/TypeScript SDK is planned.

## Install

```bash
pip install collieai
```

## Quick start

### Construct the client

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

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

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

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

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

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

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

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

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

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