ultrasafeai 0.1.0

UltrasafeAI Python SDK

UltrasafeAI Python SDK

Package: ultrasafeai
Base URL: https://api.us.tech/v1
Auth: X-API-KEY header

Installation

pip install ultrasafeai

Client Setup

from ultrasafeai import UltrasafeAI, AsyncUltrasafeAI

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

# Asynchronous
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_stream(...)
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_stream(
    model="usf-mini",
    messages=[{"role": "user", "content": "Tell me a joke"}]
):
    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_stream(
        model="usf-mini",
        messages=[{"role": "user", "content": "Tell me a joke"}]
    ):
        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_stream(
    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"}}
            ]
        }
    ]
):
    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'

import asyncio
from ultrasafeai.ultrasafeai.audio.stream import StreamClient, ConnectOptions

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

    session = await client.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
        )
    )

    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}"))

    # 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())

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

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

TranscriptEvent shape:

{
    "type": "transcript",
    "request_id": "req_abc",
    "is_final": True,
    "full_text": "Hello world this is a test",
    "committed_text": "Hello world",
    "segment": {
        "id": 3,
        "text": "this is a test",
        "is_final": True,
        "start": 1.2,
        "end": 2.8,
        "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