yua 0.2.0

YUA AI SDK — OpenAI-compatible Python client

yua

Official Python SDK for YUA AI. YUA AI Python SDK.

Installation

pip install yua

Quick Start

from yua import YUA

client = YUA(api_key="yua_sk_...")

# Streaming
stream = client.chat.completions.create(
    model="yua-normal",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)

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

Non-Streaming

response = client.chat.completions.create(
    model="yua-normal",
    messages=[{"role": "user", "content": "What is 2+2?"}],
)

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

Models

Model	Description
yua-fast	Fast responses, single segment
yua-normal	Balanced speed and quality (default)
yua-deep	Deep reasoning with multi-step thinking

Configuration

client = YUA(
    api_key="yua_sk_...",           # Required (or auth_provider)
    base_url="https://...",         # Default: https://api.yuaone.com
    workspace="ws_...",             # Optional workspace ID
    timeout=30.0,                   # Request timeout in seconds (default: 30)
    max_retries=2,                  # Auto-retry count (default: 2)
)

Context Manager

with YUA(api_key="yua_sk_...") as client:
    stream = client.chat.completions.create(...)
    for chunk in stream:
        ...
# Connection automatically closed

Custom Auth Provider

client = YUA(
    auth_provider=lambda: get_id_token(),
)

Streaming with YUA Events

YUA streams include extended events beyond standard text chunks.

stream = client.chat.completions.create(
    model="yua-deep",
    messages=[{"role": "user", "content": "Explain quantum computing"}],
    stream=True,
    thinking_profile="DEEP",
)

for chunk in stream:
    # Standard text content
    text = chunk.choices[0].delta.content
    if text:
        print(text, end="", flush=True)

    # YUA-specific events
    if chunk.yua_event:
        if chunk.yua_event.type == "activity":
            print(f"[Activity] {chunk.yua_event.data}")
        elif chunk.yua_event.type == "reasoning_block":
            print(f"[Reasoning] {chunk.yua_event.data}")
        elif chunk.yua_event.type == "suggestion":
            print(f"[Suggestions] {chunk.yua_event.data}")

Stream Helper Methods

stream = client.chat.completions.create(..., stream=True)

# Get full text after stream completes
full_text = stream.text_content()

# Get the final assembled message
message = stream.final_message()

Thread Management

# Create thread
thread = client.chat.threads.create()

# List threads
threads = client.chat.threads.list()

# Send message to specific thread
stream = client.chat.completions.create(
    model="yua-normal",
    messages=[{"role": "user", "content": "Continue our discussion"}],
    stream=True,
    thread_id=thread["threadId"],
)

# Update thread title
client.chat.threads.update(thread["threadId"], title="My Chat")

# Delete thread
client.chat.threads.delete(thread["threadId"])

Messages

# List messages in a thread
messages = client.chat.messages.list(thread_id=123)

# Send a message without streaming
client.chat.messages.create(
    thread_id=123,
    role="user",
    content="Hello",
)

Error Handling

from yua import APIError, AuthenticationError, RateLimitError

try:
    res = client.chat.completions.create(...)
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limited, retry later")
except APIError as e:
    print(f"{e.status} {e.code}: {e.message}")

YUA Event Types

Event	Description
stage	Processing stage changes (thinking, analyzing, answer)
token	Text token (mapped to delta.content)
final	Final assembled response (mapped to finish_reason="stop")
activity	Thinking activities (analyzing, planning, researching)
reasoning_block	Deep reasoning block content
reasoning_done	Reasoning phase completed
suggestion	Follow-up suggestions
memory	Memory commit events
answer_unlocked	Deep mode: answer text begins

Requirements

• Python >= 3.8
• httpx >= 0.24.0
• pydantic >= 2.0.0

License

MIT