solwyn 0.1.9

AI Agent Control Plane SDK — hard spending caps, automatic failover, per-agent cost attribution

Solwyn Python SDK

Budget enforcement, circuit breaking, and usage tracking for OpenAI, Anthropic, Google, and Amazon Bedrock LLM clients — plus any provider that speaks the OpenAI Chat Completions dialect (xAI, DeepSeek, Mistral, Qwen, Groq, Together, Fireworks, Perplexity, Azure OpenAI, OpenRouter, Ollama, vLLM, LM Studio, …).

Solwyn wraps your existing LLM client. Calls go directly to the provider — the SDK only reports metadata (token counts, latency, model name) to the Solwyn API. Prompts and responses never leave your application.

Installation

pip install solwyn

Optional extras pin tested provider-SDK floors — solwyn[openai] (also enables tiktoken-based token estimation), solwyn[anthropic], solwyn[google], solwyn[bedrock] (convenience only — the SDK never imports boto3), or solwyn[all]:

pip install solwyn[openai]

OpenAI-compatible endpoints (Groq, OpenRouter, vLLM, …) ride the openai extra; no extra of their own.

Quick Start

from openai import OpenAI
from solwyn import Solwyn

client = Solwyn(
    OpenAI(),
    api_key="sk_proj_...",
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
)

client.close()

Or use as a context manager:

with Solwyn(OpenAI(), api_key="sk_proj_...") as client:
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )

Providers

OpenAI

from openai import OpenAI
from solwyn import Solwyn

client = Solwyn(OpenAI(), api_key="sk_proj_...")
response = client.chat.completions.create(model="gpt-4o", messages=[...])

Anthropic

from anthropic import Anthropic
from solwyn import Solwyn

client = Solwyn(Anthropic(), api_key="sk_proj_...")
response = client.messages.create(model="claude-sonnet-4-20250514", max_tokens=1024, messages=[...])

Google Gemini

from google import genai
from solwyn import Solwyn

client = Solwyn(genai.Client(api_key="..."), api_key="sk_proj_...")
response = client.models.generate_content(model="gemini-2.0-flash", contents="Hello!")

Amazon Bedrock

Wrap a bedrock-runtime boto3 client. Solwyn intercepts the Converse API (converse / converse_stream), which works uniformly across every chat model Bedrock hosts — Anthropic Claude, Meta Llama, Mistral, Amazon Nova, Cohere, AI21, DeepSeek, and more. Auth stays entirely on your boto3 client (IAM credentials, profiles, roles, SigV4) — Solwyn never sees it.

import boto3
from botocore.config import Config
from solwyn import Solwyn

bedrock = boto3.client(
    "bedrock-runtime",
    region_name="us-east-1",
    # Recommended: let Solwyn own retries/failover instead of stacking
    # botocore's default retry layer (legacy mode retries up to 5 times).
    config=Config(retries={"total_max_attempts": 1}, read_timeout=60),
)

client = Solwyn(bedrock, api_key="sk_proj_...")
response = client.converse(
    modelId="us.anthropic.claude-3-5-sonnet-20241022-v2:0",
    messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
    inferenceConfig={"maxTokens": 1024},
)

Streaming preserves the boto3 contract (response["stream"]); usage settles from the stream's terminal metadata event:

response = client.converse_stream(
    modelId="amazon.nova-pro-v1:0",
    messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
)
for event in response["stream"]:
    ...

If you stop consuming the stream early, call response["stream"].close() (or wrap iteration in with response["stream"]:) to settle the budget reservation — the same close obligation raw boto3's EventStream has. close() settles exactly once with whatever usage was observed and is safe to call repeatedly.

Notes:

• Model identity is reported exactly as you pass it — foundation-model ids, cross-region inference profiles (us. / eu. / jp. / global. …), or full ARNs — together with the client's region, because Bedrock pricing is keyed per model and region. Prompt-cache reads/writes (including the 1h-TTL tier via usage.cacheDetails) and the latency/service pricing tier are captured for exact repricing.
• invoke_model / invoke_model_with_response_stream raise ConfigurationError instead of bypassing budget tracking: their usage is buried in a consume-once body alongside response content. Use Converse, or call the unwrapped boto3 client for deliberately untracked calls.
• boto3 has no per-call timeout override, so the failover deadline cannot shorten an in-flight Bedrock hop — set read_timeout in your botocore Config.
• Async works with aioboto3: AsyncSolwyn(client) inside async with session.client("bedrock-runtime") as client.
• Bedrock participates in cross-provider failover in both directions (e.g. Bedrock-Claude ⇄ direct Anthropic) via the same canonical translation subset as the other providers.

OpenAI-compatible providers

Point an openai.OpenAI client at any OpenAI-compatible endpoint via base_url and wrap it as usual. Solwyn detects the provider from the URL, so budgets, per-agent attribution, failover, and the cost dashboard all see the real provider (e.g. groq), not "openai":

from openai import OpenAI
from solwyn import Solwyn

client = Solwyn(
    OpenAI(base_url="https://api.groq.com/openai/v1", api_key="gsk_..."),
    api_key="sk_proj_...",
)
response = client.chat.completions.create(
    model="llama-3.3-70b-versatile",
    messages=[{"role": "user", "content": "Hello!"}],
)

Auto-detected providers:

Provider	Detected from	Streaming usage
xAI (Grok)	api.x.ai	automatic (final chunk); stream_options is never sent — xAI rejects it
DeepSeek	api.deepseek.com	include_usage injected
Mistral	api.mistral.ai	stream_options never sent (strict validation); final-chunk usage or estimate
Qwen (DashScope compat)	dashscope*.aliyuncs.com	include_usage injected
Groq	api.groq.com	include_usage injected; legacy x_groq.usage also handled
Together AI	api.together.xyz / api.together.ai	automatic (final chunk)
Fireworks	api.fireworks.ai	automatic (final chunk)
Perplexity (Sonar)	api.perplexity.ai	usage on streamed chunks; stream_options never sent
Azure OpenAI	*.openai.azure.com or AzureOpenAI client class	include_usage injected (skipped for "on your data" data_sources requests, which reject it)
OpenRouter	openrouter.ai	automatic (final chunk); stream_options is deprecated there
Ollama	localhost:11434	include_usage injected (older versions ignore it → estimate)
vLLM	localhost:8000	include_usage injected
LM Studio	localhost:1234	include_usage injected (pre-0.3.18 omits usage → estimate)
Anything else	any non-OpenAI base_url	generic openai_compatible; stream_options never sent

For endpoints auto-detection can't name (e.g. vLLM on a non-default port), pass the provider explicitly — on the constructor for the primary, or as the 4th element of a fallback spec:

client = Solwyn(
    OpenAI(base_url="http://gpu-box:8080/v1", api_key="-"),
    api_key="sk_proj_...",
    provider="vllm",
    fallback=[(OpenAI(base_url="https://openrouter.ai/api/v1", api_key="sk-or-..."), "openrouter/auto"),
              (other_client, "my-model", {}, "ollama")],
)

Token accounting. Budgets and attribution depend on accurate per-call usage, and "OpenAI-compatible" endpoints differ most in exactly that. Solwyn requests streaming usage only from providers where that's documented-safe, reads it from the final chunk where it arrives automatically, and — when a provider reports no usage at all (or reports an unparseable/zeroed block alongside real content) — falls back to a length-based estimate that is explicitly marked (token_details.is_estimated = true on the wire, plus a one-time SDK warning). Degraded accounting is loud and flagged, never silently zero.

The "never sent" entries above describe Solwyn's own injection policy. A stream_options you pass explicitly always reaches your configured provider untouched (drop-in contract); it is only stripped when a failover hop lands on a provider known to reject it.

Pricing. The SDK never computes cost. It reports the served (provider, model) verbatim — for OpenRouter that's the full model slug (e.g. anthropic/claude-sonnet-4.5) — and Solwyn Cloud's PricingService prices it. Models unknown to the catalog are surfaced as unpriced on the dashboard rather than silently costed at $0.

Failover. Compat providers participate fully in failover. Between two OpenAI-dialect providers (e.g. Groq → OpenRouter) requests pass through natively — tools, JSON mode, and streaming included (max_completion_tokens is rewritten to max_tokens for targets that need the legacy key). Per-call extra_headers/extra_query/extra_body are stripped on cross-provider hops — they're endpoint-scoped, authored for the original endpoint — though the fallback entry's own default_params versions still apply. Across dialects (e.g. Groq → Anthropic) the standard translation subset applies.

Known limitation. Circuit-breaker health, latency signals, and failover labeling key off the provider name. Two chain entries that resolve to the same name (two Azure resources, two unnamed gateways both detected as openai_compatible) share one health domain and are reported as model fallbacks of each other. For the same reason, a hop between same-name entries skips cross-provider request sanitization — stream_options stripping, the max_completion_tokens → max_tokens rewrite, and endpoint-scoped param stripping (extra_headers/extra_query/extra_body). A stream_options or gateway header you authored for the first endpoint reaches the second untouched and can 4xx there. Give distinct endpoints distinct provider identities where possible — explicit provider= on the constructor, or the 4th element of a fallback spec.

Async

from openai import AsyncOpenAI
from solwyn import AsyncSolwyn

async with AsyncSolwyn(
    AsyncOpenAI(),
    api_key="sk_proj_...",
) as client:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )

Streaming

Pass stream=True as you normally would. Solwyn wraps the stream transparently and reports usage when it completes:

stream = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}],
    stream=True,
)

for chunk in stream:
    print(chunk.choices[0].delta.content or "", end="")

Tagging Calls with Agent Runs

Wrap a unit of work with solwyn.run(name) to attribute every LLM call inside it to a single agent run. The dashboard groups cost and latency by run, so you can see "this nightly batch cost $4.20."

import solwyn
from openai import OpenAI

client = solwyn.Solwyn(OpenAI(), api_key="sk_proj_...")

with solwyn.run("nightly-batch") as run_id:
    client.chat.completions.create(model="gpt-4o", messages=[...])
    client.chat.completions.create(model="gpt-4o", messages=[...])

Works the same with async with and is safe across concurrent asyncio tasks — each task sees only its own active run. Calls made outside a solwyn.run(...) scope are still tracked; the API groups them into _auto-{sdk_instance_id}-{YYYY-MM-DD} using the event's UTC timestamp.

Do not open solwyn.run(...) inside an async generator. Python runs the consumer's async for body in the same context after a generator yield, so an inner generator scope would leak into customer code. The SDK rejects that pattern at scope entry. Open the scope in the consumer, or await the generator entirely inside an outer run scope.

Tasks created with asyncio.create_task(...) inside a run capture that task's context. If the task keeps making LLM calls after the with block exits, those calls are still attributed to the captured run id. Use asyncio.TaskGroup or await spawned tasks before leaving the scope when attribution must end with the block.

ThreadPoolExecutor

solwyn.run(...) uses Python contextvars. Context propagates across asyncio tasks, but not into ThreadPoolExecutor workers. Use solwyn.run_in_executor(...) when submitting threaded work that should keep the active run tag:

from concurrent.futures import ThreadPoolExecutor

with solwyn.run("nightly-batch"), ThreadPoolExecutor() as executor:
    future = solwyn.run_in_executor(executor, call_openai, prompt)
    result = future.result()

run_in_executor(...) returns the executor's concurrent.futures.Future, not an awaitable. In asyncio code, wrap it with asyncio.wrap_future(future). If you submit directly to an executor, wrap the callable with contextvars.copy_context().run(...) yourself.

Budget Enforcement

Set budget_mode to control spending:

client = Solwyn(
    OpenAI(),
    api_key="sk_proj_...",
    budget_mode="hard_deny",
)

Mode	Behavior
alert_only	Log a warning when budget is exceeded (default)
hard_deny	Raise BudgetExceededError and block the call

from solwyn import BudgetExceededError

try:
    response = client.chat.completions.create(model="gpt-4o", messages=[...])
except BudgetExceededError as e:
    print(f"Budget limit: ${e.budget_limit}, usage: ${e.current_usage}")

Configuration

Parameter	Env Var	Default	Description
api_key	SOLWYN_API_KEY	required	Solwyn project API key
api_url	SOLWYN_API_URL	https://api.solwyn.ai	Solwyn API endpoint
fail_open	SOLWYN_FAIL_OPEN	True	Allow LLM calls when Solwyn API is unreachable
budget_mode	SOLWYN_BUDGET_MODE	alert_only	Budget enforcement mode

Failover and routing (model=, fallback=, provider=, default_params=, selection_policy=, and the failover tuning knobs) are configured in code only — they take client objects and policies, not strings. See Provider Failover and Configuration.

Use env vars to avoid passing credentials in code:

export SOLWYN_API_KEY="sk_proj_..."

client = Solwyn(OpenAI())  # picks up from environment

Error Handling

All SDK errors inherit from SolwynError:

Exception	Raised when
BudgetExceededError	Budget exceeded in hard_deny mode
ProviderUnavailableError	Circuit breaker is open, or the failover chain is exhausted
ConfigurationError	Invalid API key format, invalid provider= override, or an untracked call surface (e.g. Bedrock invoke_model)
UntranslatableRequestError	A cross-provider failover hop cannot represent the request (structural labels only — never content)
UntranslatableModelError	No model mapping exists for a cross-provider failover hop

Provider errors (e.g., openai.RateLimitError) pass through unmodified.

Data Transparency

The SDK sends a MetadataEvent after each LLM call. This is everything it transmits:

Field	Type	Description
model	str	Model name (e.g., gpt-4o)
provider	str	Provider identifier (openai, anthropic, google, bedrock, groq, openrouter, …)
input_tokens	int	Input token count
output_tokens	int	Output token count
token_details	object	Breakdown: cached, reasoning, audio tokens; is_estimated flags length-based estimates when a provider reports no usage
latency_ms	float	Call duration in milliseconds
status	str	success, error, or budget_denied
is_model_fallback	bool	Whether the call was served by a same-provider entry in the fallback= chain after the primary model failed
sdk_instance_id	str	Per-process UUID for deduplication
timestamp	datetime	When the call completed (UTC)
agent_run_id	str | None	Run id from the active solwyn.run(...) scope, if any. When omitted, the API creates _auto-{sdk_instance_id}-{YYYY-MM-DD}
agent_run_name	str | None	Run name passed to solwyn.run(...), if any
provider_region	str | None	Cloud region of the serving endpoint (Bedrock — pricing is per model and region); omitted for other providers

The SDK never captures, logs, or transmits prompts or responses. This is enforced by structural tests and the privacy module.

Release Compatibility

Wire-contract changes are API-first: Solwyn Cloud must accept new fields and enum values before an SDK release ships them. As of v0.1.7 the Cloud API accepts the full current wire contract — the Bedrock and OpenAI-compatible provider values, provider_region, service_tier on budget confirms, token_details.is_estimated, 2048-char model identifiers, and per-event ingest dispositions. Optional fields are omitted entirely (never null) when unset, so payloads for providers that don't use them are byte-identical to earlier releases.

Requirements

Python 3.11+

Contributing

make install          # install in dev mode
make install-hooks    # install pre-commit hook
make check            # lint + format + typecheck
make test             # run unit tests

Links

• Documentation
• Solwyn Cloud — Dashboard, alerts, and analytics
• MPI.sh — LLM API pricing comparison

License

Apache 2.0 — see LICENSE for details.