Python SDK

The API is small enough to call without a library. The part that needs care is the response: POST /chat/completions always returns a server-sent event stream, and the model's answer arrives as a single event after the gateway has filtered it, not as a token feed.

What a client library would have to hide

• There is no non-streaming mode. Every chat call answers text/event-stream. The stream field is accepted by the request DTO and read by nothing, so response.json() fails.
• One content event, not deltas. The gateway buffers the provider's output, filters it, and emits {"id":"filtered-response","content":…,"finish_reason":"stop"} once. There is no choices[] and nothing to print progressively.
• A policy block is a 200. A rejected prompt terminates the stream with an event, not with an HTTP error.
• The two chat surfaces disagree about rejection. /chat/completions sends an error field; /sessions/:id/messages sends {"type":"content_rejected","reason":…} with no error field.
• Not every route takes the same credential. A cid_key_ bearer works on /chat/completions, /models and /api/v1/guardrails/detect. Sessions, image analysis and document analysis need a user JWT from POST /auth/login.

Use the REST API today

import json
import os
from typing import Any
 
import requests
 
GATEWAY = 'https://<gateway-host>'
HEADERS = {
    'Authorization': f'Bearer {os.environ["CID222_API_KEY"]}',
    'Content-Type': 'application/json',
}
 
 
class ContentBlocked(RuntimeError):
    """The prompt or the answer was refused by policy. Retrying the same text does not help."""
 
 
def chat(messages: list[dict], model: str = 'gpt-4o', provider: str | None = None) -> str:
    response = requests.post(
        f'{GATEWAY}/chat/completions',
        headers=HEADERS,
        json={'model': model, 'provider': provider, 'messages': messages},
        stream=True,
        # Detection runs before the provider call, and the answer is buffered:
        # allow a long read timeout.
        timeout=(10, 300),
    )
    response.raise_for_status()
 
    answer = ''
    for line in response.iter_lines(decode_unicode=True):
        if not line or not line.startswith('data: '):
            continue
        data = line[6:].strip()
        if data == '[DONE]':
            break
 
        event: dict[str, Any] = json.loads(data)
 
        # Prompt blocked before the provider was called, or a server-side failure.
        if 'error' in event:
            raise ContentBlocked(event['error'])
        # Answer discarded by the output filter.
        if event.get('type') == 'output_content_rejected':
            raise ContentBlocked(event.get('reason', 'response blocked'))
        # Routing, compression, token usage and warning telemetry.
        if event.get('type'):
            continue
 
        # The whole filtered answer, as one event.
        if isinstance(event.get('content'), str):
            answer = event['content']
 
    return answer
 
 
print(chat([{'role': 'user', 'content': 'Explain quantum computing in simple terms.'}]))

Scanning text needs no stream handling — the detection endpoint answers with plain JSON, and states the verdict in action rather than in the status code:

def detect(text: str, check_type: str = 'prompt') -> dict:
    response = requests.post(
        f'{GATEWAY}/api/v1/guardrails/detect',
        headers=HEADERS,
        json={'text': text, 'check_type': check_type},
        timeout=30,
    )
    response.raise_for_status()
    return response.json()
 
 
result = detect('Email me at john@example.com')
print(result['action'])      # "mask"
print(result['maskedText'])  # "Email me at [EMAIL]"

text must be 1–50,000 characters, and detectedEntities[].value is always a placeholder or [REDACTED] — the endpoint never echoes the raw PII it found.

What the library is planned to cover

These are intentions, not shipped behaviour. Nothing here is callable today.

• Async support — the same calls under asyncio, so a scan or a chat call does not block an event loop.
• Type hints — complete annotations, so an editor can complete the request fields.
• Stream handling — frame reassembly and the terminal events behind one call, exposed as a generator where that is useful.
• Automatic retry — exponential backoff on transport failures and 5xx, and no retry on a policy rejection, which would only produce the same verdict.
• Pydantic models — request and response validation against the gateway's own field names.

Limits and known gaps

• No package exists. Any cid222 distribution you find on a public index is not published by this project. Do not install it.
• No release date. Treat the planned list above as a direction, not a commitment.
• requests is synchronous. The client above holds a thread for the whole call, which is the full model latency plus filtering. Use httpx or aiohttp if that matters, following the same frame handling.
• Unknown request fields are dropped silently. The gateway's validation pipe strips them rather than rejecting them, so a misspelt field name produces a successful call that ignores your setting.
• iter_lines assumes well-formed frames. It is adequate here because the gateway emits one data: line per frame, but a client that must survive partial frames should buffer the raw content and split on the blank line itself.

Related pages

• Integration examples — the same client as a step-by-step procedure, plus Node.js, cURL and React.
• Error handling — status codes, the three error body shapes, and the rejection events.
• JavaScript SDK — the same situation on the JavaScript side.