rheonic-sdk 0.1.0b13

Python SDK for Rheonic observability and protect preflight enforcement.

Rheonic Python SDK

Rheonic captures provider telemetry and applies protect preflight decisions before provider calls.

Install

pip install rheonic-sdk --pre

Required environment variables

RHEONIC_INGEST_KEY=<your_project_ingest_key>
RHEONIC_BASE_URL=<value_shown_in_dashboard>

Instrument provider calls

Wrap your provider SDK once.

Telemetry is captured automatically after each provider call.

Enforcement follows Project mode in the dashboard (Observe / Protect).

OpenAI:

import json
import os
from openai import OpenAI
from rheonic import create_client, instrument_openai, RHEONICBlockedError

rheonic = create_client(
    base_url=os.environ["RHEONIC_BASE_URL"],
    ingest_key=os.environ["RHEONIC_INGEST_KEY"],
)
openai_client = instrument_openai(
    OpenAI(api_key=os.environ["OPENAI_API_KEY"]),
    client=rheonic,
    endpoint="/chat/completions",
    feature="assistant",
)

try:
    openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "hello"}],
        max_tokens=256,
    )
except RHEONICBlockedError as error:
    print(json.dumps({
        "reason": error.reason,
        "retry_after_seconds": error.retry_after_seconds,
        "blocked_until": error.blocked_until,
        "trace_id": error.trace_id,
        "request_id": error.request_id,
    }, indent=2))

Anthropic:

import json
import os
from anthropic import Anthropic
from rheonic import create_client, instrument_anthropic, RHEONICBlockedError

rheonic = create_client(
    base_url=os.environ["RHEONIC_BASE_URL"],
    ingest_key=os.environ["RHEONIC_INGEST_KEY"],
)
anthropic_client = instrument_anthropic(
    Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"]),
    client=rheonic,
    endpoint="/v1/messages",
    feature="assistant",
)

try:
    anthropic_client.messages.create(
        model="claude-3-5-sonnet-latest",
        max_tokens=256,
        messages=[{"role": "user", "content": "hello"}],
    )
except RHEONICBlockedError as error:
    print(json.dumps({
        "reason": error.reason,
        "retry_after_seconds": error.retry_after_seconds,
        "blocked_until": error.blocked_until,
        "trace_id": error.trace_id,
        "request_id": error.request_id,
    }, indent=2))

Google:

import json
import os
from google import genai
from google.genai import types
from rheonic import create_client, instrument_google, RHEONICBlockedError

rheonic = create_client(
    base_url=os.environ["RHEONIC_BASE_URL"],
    ingest_key=os.environ["RHEONIC_INGEST_KEY"],
)
google_client = instrument_google(
    genai.Client(api_key=os.environ["GOOGLE_API_KEY"]),
    client=rheonic,
    endpoint="/v1beta/models/generateContent",
    feature="assistant",
)

try:
    google_client.models.generate_content(
        model="gemini-1.5-pro",
        contents="hello",
        config=types.GenerateContentConfig(max_output_tokens=256),
    )
except RHEONICBlockedError as error:
    print(json.dumps({
        "reason": error.reason,
        "retry_after_seconds": error.retry_after_seconds,
        "blocked_until": error.blocked_until,
        "trace_id": error.trace_id,
        "request_id": error.request_id,
    }, indent=2))

RHEONICBlockedError.reason is meant to be operator-relevant. The main values are:

• tok_cap_breach
• req_cap_breach
• cooldown_active
• fail_closed

If Protect is fail_open, timeout or availability problems stay internal and the provider call continues. If Protect is fail_closed, the SDK raises RHEONICBlockedError with the feedback fields shown above.

Keep one long-lived SDK client per app process. Initialize it during app startup and reuse it for all capture and instrumentation calls so Rheonic can avoid repeated protect cold-start latency.

Optional: custom event capture

Use this only if you can't instrument a provider SDK or need custom events.

import os
from rheonic import create_client, build_event

client = create_client(
    base_url=os.environ["RHEONIC_BASE_URL"],
    ingest_key=os.environ["RHEONIC_INGEST_KEY"],
)

client.capture_event(
    build_event(
        provider="openai",
        model="gpt-4o-mini",
        request={"endpoint": "/chat/completions", "feature": "assistant", "token_explosion_tokens": 64},
        response={"total_tokens": 64, "latency_ms": 120, "http_status": 200},
    )
)

token_explosion_tokens is optional. Set it only for custom/manual events when you want token-explosion detection to use the same request-context signal that the SDK instrumentation sends to both protect and ingest.

Reference

Full quickstart:

• https://beta.rheonic.dev/quickstart

Support

[e-mail]