ultrasafeai 1.1.1

UltrasafeAI REST API with comprehensive endpoints for AI services

UltrasafeAI Python SDK

The official Python SDK for the UltrasafeAI API. Provides access to chat completions, vision, embeddings, reranking, image generation, speech-to-text, text-to-speech, real-time audio streaming, vector stores, assistants, threads, and more.

Both a synchronous client (UltrasafeAI) and an async client (AsyncUltrasafeAI) are included — the async client is a drop-in for asyncio / FastAPI workloads.

Requires Python 3.8+.

Base URL: https://api.us.tech/v1

Installation

pip install ultrasafeai

Client Setup

The client reads ULTRASAFEAI_API_KEY from the environment automatically if api_key is not passed.

from ultrasafeai import UltrasafeAI, AsyncUltrasafeAI

# Synchronous
client = UltrasafeAI(api_key="YOUR_API_KEY")

# Asynchronous (for asyncio / FastAPI)
client = AsyncUltrasafeAI(api_key="YOUR_API_KEY")

Options:

Parameter	Type	Description
api_key	str	Your UltrasafeAI API key
base_url	str	Override the base URL
timeout	float	Request timeout in seconds (default: 60)
max_retries	int	Max retry attempts
httpx_client	httpx.Client	Custom HTTP client

---

Chat Completions

Non-Streaming

Method: client.chat.completions.create(...)
Endpoint: POST /chat/completions

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

response = client.chat.completions.create(
    model="usf-mini",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Hello!"}
    ]
)

print(response.choices[0].message.content)

Async:

import asyncio
from ultrasafeai import AsyncUltrasafeAI

client = AsyncUltrasafeAI(api_key="YOUR_API_KEY")

async def main():
    response = await client.chat.completions.create(
        model="usf-mini",
        messages=[{"role": "user", "content": "Hello!"}]
    )
    print(response.choices[0].message.content)

asyncio.run(main())

Payload:

Parameter	Type	Required	Description
model	str	Yes	Model ID (e.g. usf-mini, usf-mini-x1)
messages	list	Yes	Conversation history. Roles: system, user, assistant, tool
tools	list	No	Function or custom tools the model may call
tool_choice	str | dict	No	"none", "auto", "required", or a specific tool
parallel_tool_calls	bool	No	Allow parallel tool calls (default: True)
web_search	bool	No	Enable web search (default: False)
response_format	dict	No	{"type": "text"}, {"type": "json_object"}, or {"type": "json_schema", "json_schema": {...}}
max_tokens	int	No	Max tokens to generate
temperature	float	No	Sampling temperature 0–2
top_p	float	No	Nucleus sampling probability mass
n	int	No	Number of completions to generate
stop	str | list[str]	No	Stop sequences (up to 4)
presence_penalty	float	No	Penalty for repeated tokens (-2.0 to 2.0)
frequency_penalty	float	No	Frequency-based penalty (-2.0 to 2.0)
seed	int	No	Seed for deterministic sampling
store	bool	No	Store conversation for retrieval
conversation_id	str	No	Continue an existing stored conversation
user	str	No	Stable end-user identifier

Response: ChatCompletion

{
    "id": "chatcmpl-abc123",
    "object": "chat.completion",
    "created": 1700000000,
    "model": "usf-mini",
    "conversation_id": "conv_xyz",          # present when store=True
    "choices": [
        {
            "index": 0,
            "message": {
                "role": "assistant",
                "content": "Hello! How can I help you?",
                "tool_calls": None,         # list of tool calls when finish_reason="tool_calls"
                "refusal": None
            },
            "finish_reason": "stop"         # "stop", "length", "tool_calls", "content_filter"
        }
    ],
    "usage": {
        "prompt_tokens": 12,
        "completion_tokens": 10,
        "total_tokens": 22
    }
}

---

Streaming

Method: client.chat.completions.create(...) with stream=True
Endpoint: POST /chat/completions (with stream: true)

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

for chunk in client.chat.completions.create(
    model="usf-mini",
    messages=[{"role": "user", "content": "Tell me a joke"}],
    stream=True,
):
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Async:

import asyncio
from ultrasafeai import AsyncUltrasafeAI

client = AsyncUltrasafeAI(api_key="YOUR_API_KEY")

async def main():
    async for chunk in await client.chat.completions.create(
        model="usf-mini",
        messages=[{"role": "user", "content": "Tell me a joke"}],
        stream=True,
    ):
        if chunk.choices and chunk.choices[0].delta.content:
            print(chunk.choices[0].delta.content, end="", flush=True)

asyncio.run(main())

Payload: Same as non-streaming (minus store/conversation_id not affecting stream behavior).

Response: Iterator[ChatCompletionChunk]

Each chunk:

{
    "id": "chatcmpl-abc123",
    "object": "chat.completion.chunk",
    "created": 1700000000,
    "model": "usf-mini",
    "choices": [
        {
            "index": 0,
            "delta": {
                "role": "assistant",        # only on first chunk
                "content": "Hello",         # incremental text; concatenate across chunks
                "reasoning_content": None,  # chain-of-thought when available
                "tool_calls": None          # incremental tool call data
            },
            "finish_reason": None           # non-null only on final chunk
        }
    ],
    "usage": None  # present only on last chunk when stream_options.include_usage=True
}

---

Vision

Vision uses the same chat.completions.create / create_stream methods. Pass a list of content parts instead of a plain string for the content field of a user message.

Non-Streaming

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

response = client.chat.completions.create(
    model="usf-mini",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What is in this image?"},
                {
                    "type": "image_url",
                    "image_url": {"url": "https://example.com/image.jpg"}
                }
            ]
        }
    ]
)

print(response.choices[0].message.content)

Base64 image:

import base64

with open("image.jpg", "rb") as f:
    b64 = base64.b64encode(f.read()).decode()

response = client.chat.completions.create(
    model="usf-mini",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "Describe this image."},
                {
                    "type": "image_url",
                    "image_url": {"url": f"data:image/jpeg;base64,{b64}"}
                }
            ]
        }
    ]
)

Streaming

for chunk in client.chat.completions.create(
    model="usf-mini",
    messages=[
        {
            "role": "user",
            "content": [
                {"type": "text", "text": "What's in this image?"},
                {"type": "image_url", "image_url": {"url": "https://example.com/image.jpg"}}
            ]
        }
    ],
    stream=True,
):
    if chunk.choices and chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="", flush=True)

Content part types:

Type	Fields	Description
text	text: str	Plain text content
image_url	image_url: {url: str}	URL or data:image/...;base64,... string

Response: Same ChatCompletion / ChatCompletionChunk as standard chat completions.

---

Embeddings

Method: client.embeddings.create(...)
Endpoint: POST /embeddings

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

# Single string
response = client.embeddings.create(
    model="usf-embed",
    input="The quick brown fox"
)

print(response.data[0].embedding)  # list of floats

# Multiple strings
response = client.embeddings.create(
    model="usf-embed",
    input=["First sentence", "Second sentence"],
    dimensions=512
)

Payload:

Parameter	Type	Required	Description
model	str	Yes	Embedding model ID (e.g. usf-embed)
input	str | list[str] | list[int] | list[list[int]]	Yes	Text or token arrays to embed. Max 8192 tokens per input, 300k tokens total
dimensions	int	No	Output embedding dimensions (supported on usf-embed and later)
encoding_format	str	No	"float" (default) or "base64"
user	str	No	End-user identifier

Response: EmbeddingResponse

{
    "object": "list",
    "data": [
        {
            "object": "embedding",
            "index": 0,
            "embedding": [0.0023, -0.0142, ...]  # list of floats
        }
    ],
    "model": "usf-embed",
    "usage": {
        "prompt_tokens": 8,
        "total_tokens": 8
    }
}

---

Reranker

Method: client.rerank.create(...)
Endpoint: POST /rerank

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

response = client.rerank.create(
    model="usf-rerank",
    query="What is machine learning?",
    texts=[
        "Machine learning is a subset of AI.",
        "The weather is sunny today.",
        "Deep learning uses neural networks."
    ],
    top_n=2
)

for result in response.results:
    print(result.index, result.relevance_score, result.text)

Payload:

Parameter	Type	Required	Description
model	str	Yes	Rerank model ID (e.g. usf-rerank)
query	str	Yes	Search query to rank documents against
texts	list[str]	Yes	Documents to rerank
top_n	int	No	Number of top results to return

Response: CreateRerankResponse

{
    "results": [
        {
            "index": 0,
            "relevance_score": 0.97,
            "text": "Machine learning is a subset of AI."
        },
        {
            "index": 2,
            "relevance_score": 0.85,
            "text": "Deep learning uses neural networks."
        }
    ]
}

---

Image Generation

Generate

Method: client.images.generate(...)
Endpoint: POST /images/generations

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

response = client.images.generate(
    model="usf-mini-image",
    prompt="A futuristic city at sunset",
    size="1024x1024",
    n=1,
    response_format="url"
)

print(response.data[0].url)

Payload:

Parameter	Type	Required	Description
model	str	Yes	Image model ID (e.g. usf-mini-image)
prompt	str	Yes	Text description of the image to generate
size	str	No	"256x256", "512x512", "1024x1024"
n	int	No	Number of images to generate
response_format	str	No	"url" (default) or "b64_json"

Response: ImageResponse

{
    "created": 1700000000,
    "data": [
        {"url": "https://..."},     # when response_format="url"
        {"b64_json": "iVBORw..."}   # when response_format="b64_json"
    ]
}

Edit Image

Method: client.images.edit_image(...)
Endpoint: POST /images/edits

with open("image.png", "rb") as img, open("mask.png", "rb") as msk:
    response = client.images.edit_image(
        image=img,
        mask=msk,
        prompt="Add a rainbow to the sky",
        model="usf-mini-image"
    )

print(response.data[0].url)

Payload:

Parameter	Type	Required	Description
image	File	No	Base image file
mask	File	No	Mask PNG (transparent areas are edited)
prompt	str	No	Edit instruction
model	str	No	Model ID

Create Variations

Method: client.images.create_image_variations(...)
Endpoint: POST /images/variations

with open("image.png", "rb") as img:
    response = client.images.create_image_variations(
        image=img,
        model="usf-mini-image"
    )

---

Speech to Text

Method: client.audio.transcriptions.create(...)
Endpoint: POST /audio/transcribe

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

with open("audio.mp3", "rb") as f:
    response = client.audio.transcriptions.create(
        file=f,
        model="usf-mini-asr",
        language="en",
        response_format="json"
    )

print(response.text)

Payload:

Parameter	Type	Required	Description
file	File	Yes	Audio file (mp3, mp4, wav, flac, ogg, webm, etc.)
model	str	Yes	ASR model ID (e.g. usf-mini-asr)
language	str	No	ISO 639-1 language code (e.g. "en", "es")
response_format	str	No	"json" (default), "text", "srt", "verbose_json", "vtt"

Response: TranscriptionResponse

{
    "text": "Hello, this is a transcription.",
    "language": "en",
    "duration": 3.5
}

---

Live ASR (WebSocket)

Live ASR uses a WebSocket-based client separate from the main HTTP client.

Class: StreamClient
Endpoint: wss://api.us.tech/v1/audio/stream

Install dependency: pip install 'websockets>=10'

The realtime client is wired onto the main client as client.audio.stream, so no separate API key is needed. ConnectOptions lives in ultrasafeai.audio_stream.realtime.

import asyncio
from ultrasafeai import UltrasafeAI
from ultrasafeai.audio_stream.realtime import ConnectOptions

async def main():
    client = UltrasafeAI(api_key="YOUR_API_KEY")

    session = await client.audio.stream.connect(
        ConnectOptions(
            model="usf-mini-asr",
            sample_rate=16000,
            audio_format="pcm_s16le",
            enable_vad=False,
            partial_results=True,
            interim_min_duration_ms=500,
            full_context_retranscription=True,
        ),
        max_retries=3,   # initial-connect retries with exponential backoff + jitter
        backoff_ms=500,
    )

    # Handlers receive a typed `TranscriptEvent` (pydantic model — attribute access).
    session.on("ready", lambda e: print("Connected — streaming audio"))
    session.on("transcript", lambda e: print(e.full_text))
    session.on("close", lambda code, reason: print(f"Closed: {code} {reason}"))
    session.on("parse_error", lambda exc, raw: print(f"Bad frame: {exc}"))

    # Send PCM audio frames
    with open("audio.raw", "rb") as f:
        while chunk := f.read(4096):
            await session.send(chunk)

    await session.close()

asyncio.run(main())

You can also construct the client directly:

from ultrasafeai.audio_stream.realtime import StreamClient, ConnectOptions

client = StreamClient(api_key="YOUR_API_KEY")
session = await client.connect(ConnectOptions(model="usf-mini-asr"))

ConnectOptions parameters:

Parameter	Type	Default	Description
model	str	"usf-mini-asr"	ASR model ID
sample_rate	int	16000	Audio sample rate in Hz
audio_format	str	"pcm_s16le"	"pcm_s16le" or "pcm_f32le"
enable_vad	bool	False	Enable voice activity detection
partial_results	bool	True	Emit partial results before segment is final
interim_min_duration_ms	int	500	Min audio duration (ms) before emitting interim
full_context_retranscription	bool	True	Re-transcribe with full audio context for accuracy

connect() retry options (keyword args, parity with the TypeScript client):

Parameter	Type	Default	Description
max_retries	int	3	Initial-connect retry attempts
backoff_ms	float	500	Base backoff (ms); doubles each attempt, with jitter

Session events:

Event	Handler signature	Description
ready	(event: TranscriptEvent)	Server ready to receive audio
transcript	(event: TranscriptEvent)	Transcription result (partial or final)
speech_activity	(event: TranscriptEvent)	VAD speech start/end
control	(event: TranscriptEvent)	Lifecycle signal (action: "stop")
error	(event: TranscriptEvent)	Server-side error
close	(code: int, reason: str)	Connection closed
ws_error	(exc: Exception)	WebSocket error
parse_error	(exc: Exception, raw: str)	A frame could not be decoded/parsed (surfaced, not swallowed)

TranscriptEvent is the Fern-generated pydantic model (ultrasafeai.types.TranscriptEvent) — access fields as attributes:

e.type             # "transcript"
e.request_id       # "req_abc"
e.is_final         # True
e.full_text        # "Hello world this is a test"
e.committed_text   # "Hello world"
e.segment.text     # "this is a test"   (e.segment is a TranscriptSegment)
e.segment.confidence  # 0.95

---

Vector Stores

Access: client.vector_stores

Create Vector Store

Method: client.vector_stores.create(...)
Endpoint: POST /vector_stores

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

with open("document.pdf", "rb") as f:
    store = client.vector_stores.create(
        name="My Knowledge Base",
        files=[f]
    )

print(store.id)      # e.g. "vs_abc123"
print(store.status)  # poll until "ready"

Payload:

Parameter	Type	Required	Description
name	str	Yes	Display name for the vector store
files	list[File]	No	Files to upload and index immediately

Response: VectorStore

{
    "id": "vs_abc123",
    "object": "vector_store",
    "created_at": 1700000000,
    "name": "My Knowledge Base",
    "status": "ready",
    "file_counts": {"in_progress": 0, "completed": 1, "failed": 0, "cancelled": 0, "total": 1}
}

List Vector Stores

response = client.vector_stores.list(limit=20)
for store in response.data:
    print(store.id, store.name, store.status)

Retrieve Vector Store

store = client.vector_stores.retrieve(vector_store_id="vs_abc123")
print(store.status)

Delete Vector Store

result = client.vector_stores.delete(vector_store_id="vs_abc123")
print(result.deleted)  # True

Search Vector Store

Method: client.vector_stores.search(vector_store_id, ...)
Endpoint: POST /vector_stores/{vector_store_id}/search

results = client.vector_stores.search(
    vector_store_id="vs_abc123",
    query="What is the refund policy?"
)

for item in results.data:
    print(item)

File Management

Upload File to Vector Store

with open("doc.pdf", "rb") as f:
    file = client.vector_stores.upload_file(vector_store_id="vs_abc123", file=f)

List Vector Store Files

files = client.vector_stores.list_files(vector_store_id="vs_abc123", limit=20)
for f in files.data:
    print(f.id)

Retrieve / Delete Vector Store File

file = client.vector_stores.retrieve_file(vector_store_id="vs_abc123", file_id="file_xyz")

result = client.vector_stores.delete_file(vector_store_id="vs_abc123", file_id="file_xyz")
print(result.deleted)  # True

File Batches

# Create a batch of files by ID
batch = client.vector_stores.create_file_batch(
    vector_store_id="vs_abc123",
    file_ids=["file_abc", "file_def"]
)

# Retrieve batch status
status = client.vector_stores.retrieve_file_batch(
    vector_store_id="vs_abc123",
    batch_id=batch.id
)

# Cancel a running batch
client.vector_stores.cancel_file_batch(vector_store_id="vs_abc123", batch_id=batch.id)

# List files in a batch
batch_files = client.vector_stores.list_batch_files(
    vector_store_id="vs_abc123",
    batch_id=batch.id
)

---

Assistants

Access: client.assistants

Create Assistant

assistant = client.assistants.create(
    model="usf-mini",
    name="My Assistant",
    description="A helpful customer support bot",
    instructions="You are a customer support agent. Be concise and friendly.",
    tools=[{"type": "code_interpreter"}],
    temperature=0.5
)

print(assistant.id)

Payload:

Parameter	Type	Required	Description
model	str	Yes	Model ID
name	str	No	Assistant name
description	str	No	Short description
instructions	str	No	System prompt / instructions
tools	list[dict]	No	Tool definitions (e.g. [{"type": "code_interpreter"}])
tool_resources	dict	No	Resources for tools
metadata	dict	No	Arbitrary key-value metadata
temperature	float	No	Sampling temperature
top_p	float	No	Nucleus sampling
response_format	str	No	Response format

Response: Assistant

{
    "id": "asst_abc123",
    "object": "assistant",
    "created_at": 1700000000,
    "name": "My Assistant",
    "description": "A helpful customer support bot",
    "model": "usf-mini",
    "instructions": "You are a customer support agent.",
    "tools": [{"type": "code_interpreter"}]
}

List Assistants

assistants = client.assistants.list(limit=20)
for asst in assistants.data:
    print(asst.id, asst.name)

Payload:

Parameter	Type	Description
limit	int	Max items to return
after	str	Pagination cursor

Retrieve Assistant

assistant = client.assistants.retrieve(assistant_id="asst_abc123")
print(assistant.name)

Delete Assistant

result = client.assistants.delete(assistant_id="asst_abc123")
print(result.deleted)  # True

Response: DeletedResponse

{"id": "asst_abc123", "object": "assistant.deleted", "deleted": True}

---

Threads

Access: client.threads

Create Thread

Method: client.threads.create(...)
Endpoint: POST /threads

from ultrasafeai import UltrasafeAI

client = UltrasafeAI(api_key="YOUR_API_KEY")

thread = client.threads.create(
    messages=[
        {"role": "user", "content": "Hello, I need help with my account."}
    ]
)

print(thread.id)  # e.g. "thread_abc123"

Payload:

Parameter	Type	Required	Description
messages	list[dict]	No	Initial messages to seed the thread
metadata	dict	No	Arbitrary key-value metadata

Response: Thread

{
    "id": "thread_abc123",
    "object": "thread",
    "created_at": 1700000000,
    "metadata": {}
}

List Threads

threads = client.threads.list(limit=20)
for t in threads.data:
    print(t.id, t.created_at)

Retrieve Thread

thread = client.threads.retrieve(thread_id="thread_abc123")
print(thread.id)

---

Thread Messages

Thread messages are managed via client.threads.add_message and client.threads.list_messages.

Add Message to Thread

Method: client.threads.add_message(thread_id, ...)
Endpoint: POST /threads/{thread_id}/messages

message = client.threads.add_message(
    thread_id="thread_abc123",
    role="user",
    content="Can you summarize my previous question?"
)

print(message.id)    # e.g. "msg_xyz"
print(message.role)  # "user"

Payload:

Parameter	Type	Required	Description
role	str	Yes	Message role: "user" or "assistant"
content	str	Yes	Text content of the message
attachments	list[dict]	No	File attachments
metadata	dict	No	Arbitrary key-value metadata

Response: Message

{
    "id": "msg_xyz",
    "object": "thread.message",
    "created_at": 1700000000,
    "thread_id": "thread_abc123",
    "role": "user",
    "content": [{"type": "text", "text": {"value": "Can you summarize my previous question?"}}]
}

List Messages in Thread

Method: client.threads.list_messages(thread_id, ...)
Endpoint: GET /threads/{thread_id}/messages

messages = client.threads.list_messages(thread_id="thread_abc123", limit=20)
for msg in messages.data:
    print(msg.role, msg.content)

Run Thread with Assistant

Method: client.threads.run(thread_id, ...)
Endpoint: POST /threads/{thread_id}/runs

run = client.threads.run(
    thread_id="thread_abc123",
    assistant_id="asst_abc123",
    model="usf-mini",
    instructions="Be concise."
)

print(run.id)      # e.g. "run_abc"
print(run.status)  # "queued" | "in_progress" | "completed" | "failed"

Payload:

Parameter	Type	Required	Description
assistant_id	str	Yes	Assistant to use for this run
model	str	No	Override the assistant's model
instructions	str	No	Override the assistant's instructions
tools	list[dict]	No	Override the assistant's tools
metadata	dict	No	Arbitrary key-value metadata

---

Models

Access: client.models

List Models

Method: client.models.list()
Endpoint: GET /models

response = client.models.list()
for model in response.data:
    print(model.id, model.type, model.description)

Response: ListModelsResponse

{
    "object": "list",
    "data": [
        {
            "id": "usf-mini",
            "object": "model",
            "name": "USF Mini",
            "type": "chat",
            "description": "Fast and efficient chat model",
            "is_active": True,
            "created": 1700000000,
            "owned_by": "ultrasafeai"
        }
    ]
}

Retrieve Model

Method: client.models.retrieve(model)
Endpoint: GET /models/{model}

model = client.models.retrieve("usf-mini")
print(model.id, model.is_active)

Response: Model

{
    "id": "usf-mini",
    "object": "model",
    "name": "USF Mini",
    "type": "chat",
    "description": "Fast and efficient chat model",
    "is_active": True,
    "created": 1700000000,
    "owned_by": "ultrasafeai"
}

---

Error Handling

from ultrasafeai.errors import UnauthorizedError, BadRequestError, PaymentRequiredError

try:
    response = client.chat.completions.create(
        model="usf-mini",
        messages=[{"role": "user", "content": "Hello"}]
    )
except UnauthorizedError as e:
    print("Invalid API key:", e)
except BadRequestError as e:
    print("Bad request:", e)
except PaymentRequiredError as e:
    print("Insufficient credits:", e)

Exception	HTTP Status	Description
UnauthorizedError	401	Invalid or missing API key
BadRequestError	400	Invalid request parameters
PaymentRequiredError	402	Insufficient account credits

---

Retries

The client automatically retries on connection errors, timeouts, and 429/5xx responses with exponential backoff. Default is 2 retries.

# Disable retries
client = UltrasafeAI(api_key="...", max_retries=0)

# Increase retries
client = UltrasafeAI(api_key="...", max_retries=5)

# Override per request
client.chat.completions.create(
    model="usf-mini",
    messages=[...],
    request_options={"max_retries": 0},
)

---

Timeouts

Requests time out after 60 seconds by default.

# Set globally
client = UltrasafeAI(api_key="...", timeout=30.0)

# Override per request
client.chat.completions.create(
    model="usf-mini",
    messages=[...],
    request_options={"timeout_in_seconds": 10},
)

---

Lib Helpers

The SDK ships ergonomics helpers under ultrasafeai.lib.

---

Streaming Accumulator

Import: from ultrasafeai.lib.streaming.chat import ChatCompletionStream

Wraps a raw stream=True response and accumulates deltas so you can read the final assembled message after iteration.

import asyncio
from ultrasafeai import AsyncUltrasafeAI
from ultrasafeai.lib.streaming.chat import ChatCompletionStream

client = AsyncUltrasafeAI()  # reads ULTRASAFEAI_API_KEY from environment

async def main():
    raw = await client.chat.completions.create(
        model="usf-mini",
        messages=[{"role": "user", "content": "Count from 1 to 5."}],
        stream=True,
    )
    stream = ChatCompletionStream(raw)

    async for chunk in stream:
        delta = chunk.choices[0].delta.content if chunk.choices else None
        if delta:
            print(delta, end="", flush=True)
    print()

    completion = stream.get_final_completion()
    print(completion["choices"][0]["finish_reason"])       # "stop"
    print(completion["choices"][0]["message"]["content"])  # full assembled text

asyncio.run(main())

Methods:

Method	Description
ChatCompletionStream(stream)	Wrap a raw async iterable of chunks
async for chunk in stream	Iterate and accumulate simultaneously
await stream.until_done()	Drain the stream without iterating manually, returns self
stream.get_final_completion()	Returns assembled dict with choices — call after stream is consumed

---

Tool Helpers

Import: from ultrasafeai.lib._tools import pydantic_function_tool, normalize_tools_for_api

from pydantic import BaseModel, Field
from ultrasafeai import UltrasafeAI
from ultrasafeai.lib._tools import pydantic_function_tool, normalize_tools_for_api

client = UltrasafeAI()

class GetWeather(BaseModel):
    """Get the current weather for a city."""
    city: str = Field(description="City name")
    unit: str = Field(default="celsius", description="celsius or fahrenheit")

weather_tool = pydantic_function_tool(GetWeather)

response = client.chat.completions.create(
    model="usf-mini",
    messages=[{"role": "user", "content": "What's the weather in London?"}],
    tools=[weather_tool],
)

# Parse the tool call result back into the Pydantic model
import json
tc = response.choices[0].message.tool_calls[0]
args = weather_tool.model(**json.loads(tc.function.arguments))
print(args.city, args.unit)

API accepts flat format (name/description/parameters at top level). If you have tools in OpenAI nested format ({"type": "function", "function": {...}}), normalise them first:

from ultrasafeai.lib._tools import normalize_tools_for_api

tools = normalize_tools_for_api([
    {"type": "function", "function": {"name": "get_weather", "description": "...", "parameters": {...}}}
])
# → [{"name": "get_weather", "description": "...", "parameters": {...}}]

Functions:

Function	Description
pydantic_function_tool(model, *, name?, description?)	Build a flat tool from a Pydantic model with strict JSON schema. Returns PydanticFunctionTool (dict subclass) with .model for response parsing
normalize_tool_for_api(tool)	Convert a flat or OpenAI-nested tool to flat
normalize_tools_for_api(tools)	Normalize a list of tools to flat
tool_message(*, tool_call_id, name, content)	Build a tool role message dict for the conversation

---

Structured Output

Import: from ultrasafeai.lib._parsing import parse_chat_completion

Parse a completion's JSON content directly into a Pydantic model.

from pydantic import BaseModel
from ultrasafeai import UltrasafeAI
from ultrasafeai.lib._parsing import parse_chat_completion

client = UltrasafeAI()

class Sentiment(BaseModel):
    sentiment: str   # "positive" | "neutral" | "negative"
    confidence: float

raw = client.chat.completions.create(
    model="usf-mini",
    messages=[
        {"role": "system", "content": "Respond with JSON only."},
        {"role": "user", "content": "The product is excellent!"},
    ],
    response_format={"type": "json_object"},
)

result = parse_chat_completion(raw, Sentiment)
print(result.parsed.sentiment)   # "positive"
print(result.parsed.confidence)  # 0.97

API: parse_chat_completion(completion, response_format) — parses choices[0].message.content as JSON, validates against the Pydantic model, and returns ParsedChatCompletion[T] with a .parsed attribute containing the typed instance. All other completion attributes (id, model, usage, etc.) are forwarded transparently.

---

Pagination

Import: from ultrasafeai.lib.pagination import SyncPage, SyncCursorPage

List endpoints return SyncPage[T] or SyncCursorPage[T]. Both are directly iterable.

from ultrasafeai import UltrasafeAI

client = UltrasafeAI()

page = client.models.list()

# Iterate directly
for model in page:
    print(model.id)

# Or access .data
print(page.data[0].id)
print(len(page))

Classes:

Class	Fields	Description
SyncPage[T]	data: list[T], object: str	Standard list response, directly iterable
SyncCursorPage[T]	data, has_more, next_cursor?	Cursor-paginated list, directly iterable