ruoyi-ai-sdk 0.3.0

Python SDK for the RuoYi AI engine (OpenAI-compatible chat, embeddings, rerank, agent forward, traces, providers).

ruoyi-ai-sdk

Python SDK for the RuoYi AI engine.

A 1:1 Python port of the Java ruoyi-ai-sdk. Wraps the same OpenAPI surface (models, embeddings, rerank, agent forward, traces, providers, chat history) and exposes both synchronous and asynchronous APIs.

• PyPI: ruoyi-ai-sdk
• Import: ruoyi_ai_sdk
• Python: 3.9+
• HTTP: httpx (sync + async)
• Models: pydantic v2

Install

pip install ruoyi-ai-sdk

Optional extras:

pip install ruoyi-ai-sdk[nacos]   # Nacos discovery via nacos-sdk-python (when available)
pip install ruoyi-ai-sdk[sse]     # unused; SDK ships its own SSE parser
pip install ruoyi-ai-sdk[dev]     # pytest, respx, ruff, freezegun, ...

Quickstart

from ruoyi_ai_sdk import EngineClient
from ruoyi_ai_sdk.models.openai import ChatRequest, ChatMessage

client = (
    EngineClient.builder()
    .base_url("https://your-engine.example.com")
    .api_key("sk-xxxxxxxx")
    .timeout(15.0)
    .logging(enabled=True)
    .build()
)

info = client.health.check()
print(info.status)

resp = client.chat.block(
    ChatRequest(
        model="gpt-4o-mini",
        messages=[ChatMessage.system("You are helpful."), ChatMessage.user("Hi!")],
    )
)
print(resp.choices[0].message.content)

client.close()

Async API

import asyncio
from ruoyi_ai_sdk import AsyncEngineClient
from ruoyi_ai_sdk.models.openai import ChatRequest, ChatMessage

async def main():
    async with (
        AsyncEngineClient.builder()
        .base_url("https://your-engine.example.com")
        .api_key("sk-xxxxxxxx")
        .build() as client
    ):
        resp = await client.chat.block(
            ChatRequest(model="gpt-4o-mini", messages=[ChatMessage.user("hi")])
        )
        print(resp.choices[0].message.content)

asyncio.run(main())

Endpoints

Method	Path	Client
GET	/openapi/health	client.health.check()
GET	/openapi/chat/msg/list	client.chat_record.list_messages(query, page_num, page_size)
GET	/openapi/chat/msg/result	client.chat_record.get_message_result(msg_id, member_id)
GET	/openapi/chat/msg/clear	client.chat_record.clear_messages(member_id, session_id)
POST	/openapi/chat/msg/feedback	client.chat_record.submit_feedback(feedback)
GET	/openapi/model/providers	client.provider.list_providers()
GET	/openapi/model/list	client.provider.list_models()
GET	/openapi/model/tree	client.provider.get_tree()
GET	/openapi/trace/{traceId}	client.trace.get_chain(trace_id)
GET	/openapi/trace/page	client.trace.page(query, page_num, page_size)
POST	/openapi/proxy-model/chat/completions (block)	client.chat.block(request)
POST	/openapi/proxy-model/chat/completions (stream)	client.chat.stream(request)
POST	/openapi/proxy-model/embeddings	client.embedding.create(request)
POST	/openapi/proxy-model/rerank	client.rerank.rerank(request)
POST	/openapi/proxy-agent/chat (block)	client.agent.forward_block(app_id, body)
POST	/openapi/proxy-agent/chat (stream)	client.agent.forward_stream(app_id, body)
POST	/openapi/proxy-agent/chat (async)	client.agent.forward_async(app_id, body)

Streaming

Chat (OpenAI-compatible chunks)

for chunk in client.chat.stream(
    ChatRequest(model="gpt-4o-mini", messages=[ChatMessage.user("hi")])
):
    for choice in chunk.choices:
        if choice.delta.content:
            print(choice.delta.content, end="", flush=True)

Chat (stream + aggregate into one response)

If you want the live token flow and a single ChatResponse at the end (the same shape you'd get from chat.block(...)), use stream_block:

from ruoyi_ai_sdk.streaming.chat_accumulator import PartialContent

def on_token(ev):
    if isinstance(ev, PartialContent):
        print(ev.text, end="", flush=True)

resp = client.chat.stream_block(
    ChatRequest(model="gpt-4o-mini", messages=[ChatMessage.user("hi")]),
    on_partial=on_token,
)
print()
# resp.choices[0].message.content is the full assembled string
# resp.choices[0].finish_reason / resp.usage are also captured

The accumulator (also exported as ChatChunkAccumulator for callers that want to drive the iteration themselves) handles tool-call streaming (grouping deltas by index, joining partial arguments JSON), reasoning content (DeepSeek / o1 style), and usage on the final chunk. Mirrors the Java SDK's EventCallBackChat.

Agent forward (custom event types)

from ruoyi_ai_sdk.streaming.agent_events import AgentEvent, AgentEventAdapter

class PrintCallback(AgentEventAdapter):
    def on_thinking(self, event: AgentEvent) -> None:
        print(f"[thinking] {event.data}")
    def on_message(self, event: AgentEvent) -> None:
        print(f"[message] {event.data}")
    def on_complete(self) -> None:
        print("[done]")

for event in client.agent.forward_stream(
    "your-app-id", {"question": "hi"}, callback=PrintCallback()
):
    pass

Multi-server & discovery

from ruoyi_ai_sdk.discovery import StaticDiscovery, make_nacos_discovery_http
from ruoyi_ai_sdk.loadbalancer import RoundRobinLoadBalancer
from ruoyi_ai_sdk.circuit_breaker import CircuitBreaker

# Static round-robin across three servers:
client = (
    EngineClient.builder()
    .servers(["https://engine-a", "https://engine-b", "https://engine-c"])
    .load_balancer(RoundRobinLoadBalancer())
    .circuit_breaker(CircuitBreaker(failure_threshold=5, recovery_interval=60.0))
    .api_key("sk-xxx")
    .build()
)

# Nacos-backed discovery:
nacos = make_nacos_discovery_http(
    server_addr="http://nacos:8848",
    namespace="public",
    service_name="openapi",
)
client = (
    EngineClient.builder()
    .discovery(nacos)
    .api_key("sk-xxx")
    .build()
)

Timeouts

The SDK exposes three knobs (mirrors the Java AiTimeout constants):

Knob	Default	Used for
connect_timeout	5s	TCP / TLS handshake (httpx.Timeout(connect=...))
call_timeout	30s	read / write / pool for blocking HTTP calls
stream_timeout	120s	read / write / pool for SSE streams

client = (
    EngineClient.builder()
    .base_url("https://engine.example.com")
    .api_key("sk-xxx")
    # Pick any of:
    .timeout(15.0)            # shortcut for call_timeout
    .connect_timeout(2.0)
    .call_timeout(15.0)
    .stream_timeout(180.0)
    .build()
)

EngineTimeoutError is raised on either the connect or read deadline.

Errors

from ruoyi_ai_sdk import (
    EngineClientError,
    EngineAPIError,         # non-2xx HTTP
    EngineBusinessError,    # server-side business error (R.code != 200)
    EngineTimeoutError,
    EngineConnectionError,
    EngineConfigError,
)

try:
    client.chat.block(req)
except EngineBusinessError as e:
    print(f"server-side error {e.code}: {e.msg}")
except EngineAPIError as e:
    print(f"HTTP {e.status_code}: {e.body}")

Examples

See examples/ for runnable scripts:

1. 01_health.py — health check
2. 02_chat_block.py — blocking chat
3. 03_chat_stream.py — streaming chat
4. 04_embedding.py — embeddings
5. 05_rerank.py — rerank
6. 06_agent_block.py — agent forward (blocking)
7. 07_agent_stream.py — agent forward (streaming)
8. 08_async_full.py — async tour
9. 09_multi_server_lb.py — multi-server round-robin
10. 10_nacos_discovery.py — Nacos discovery

Development

git clone https://github.com/yanzhao/ruoyi-ai
cd ruoyi-ai/ruoyi-ai-sdk
pip install -e ".[dev]"
ruff check src tests
ruff format --check src tests
pytest -q --cov=ruoyi_ai_sdk --cov-report=term-missing

License

Apache-2.0.