AsyncSolwyn

import asyncio
import os
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn

async def main():
    async with AsyncSolwyn(
        AsyncOpenAI(),
        api_key=os.environ["SOLWYN_API_KEY"],
    ) as client:
        response = await client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": "Hello!"}],
        )
        print(response.choices[0].message.content)

asyncio.run(main())

Asynchronous client wrapper. Same API and behavior as Solwyn, but all I/O operations are async.

Constructor

AsyncSolwyn(
    client,
    *,
    api_key: str | None = None,
    model: str | None = None,
    provider: str | None = None,
    fallback: list[tuple] | None = None,
    default_params: dict | None = None,
    selection_policy: SelectionPolicy | None = None,
    **config_kwargs,
)

Parameters

Parameter	Type	Required	Description
`client`	`openai.AsyncOpenAI \| anthropic.AsyncAnthropic \| genai.Client \|` aioboto3 `bedrock-runtime` client	Yes	The async LLM provider client to wrap. Provider is auto-detected — including OpenAI-compatible endpoints via the client's `base_url`. Becomes the primary entry in the failover chain.
`api_key`	`str \| None`	Yes (kwarg or env)	Solwyn API key. Must match `sk_proj_<64 lowercase hex chars>`. Falls back to `SOLWYN_API_KEY` if omitted.
`model`	`str \| None`	No	Model for the primary entry. Optional for single-provider use; set it when configuring a fallback chain.
`provider`	`str \| None`	No	Explicit provider identity for the primary client, for endpoints auto-detection cannot name (e.g. `provider="vllm"` on a non-default port). An override relabels attribution within the client's API dialect only; an unknown or dialect-mismatched value raises `ConfigurationError` at construction.
`fallback`	`list[tuple] \| None`	No	Failover chain entries: `(client, model)`, `(client, model, default_params)`, or `(client, model, default_params, provider)` tuples, in attempt order. Use the async client classes here. See Provider Failover.
`default_params`	`dict \| None`	No	Global fill-absent request params applied to every entry (per-entry `default_params` wins).
`selection_policy`	`SelectionPolicy \| None`	No	Candidate ordering policy. Defaults to `HealthBasedPolicy`; `LatencyPolicy` is also available.
`**config_kwargs`		No	Additional configuration options. See SolwynConfig.

api_key, model, provider, fallback, default_params, and selection_policy are all keyword-only arguments. If SOLWYN_API_KEY is present in the environment, you can omit api_key:

# With SOLWYN_API_KEY set in the environment:
client = AsyncSolwyn(AsyncOpenAI())

Raises

ConfigurationError -- Invalid API key format, malformed fallback spec, unknown config field, or an invalid provider= override (unknown name, unrecognized client, or a dialect mismatch between the override and the detected client). Also raised at call time when invoke_model / invoke_model_with_response_stream is called on a Bedrock client (field="invoke_model").
UntranslatableRequestError -- (at call time) a cross-provider failover hop cannot translate the request shape.
UntranslatableModelError -- (at call time) a fallback entry has no model configured for its provider.

Example

import os
from anthropic import AsyncAnthropic
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn

client = AsyncSolwyn(
    AsyncOpenAI(),
    model="gpt-4o",
    api_key=os.environ["SOLWYN_API_KEY"],
    budget_mode="hard_deny",
    fallback=[(AsyncAnthropic(), "claude-sonnet-4-5", {"max_tokens": 1024})],
)

Methods

close()

await client.close() -> None

Shuts down the async metadata reporter and closes HTTP connections. Always call await client.close() when you are done, or use async with.

Async context manager

AsyncSolwyn supports async with. The __aenter__ method starts the background metadata reporter, and __aexit__ flushes pending events and closes connections:

import os
from openai import AsyncOpenAI
from solwyn import AsyncSolwyn

async with AsyncSolwyn(
    AsyncOpenAI(),
    api_key=os.environ["SOLWYN_API_KEY"],
) as client:
    response = await client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "Hello!"}],
    )
# close() is called automatically on exit

Call surfaces

All calls use await:

OpenAI

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

Anthropic

response = await client.messages.create(
    model="claude-sonnet-4-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

Google Gemini

response = await client.models.generate_content(
    model="gemini-2.0-flash",
    contents="Hello!",
)

Amazon Bedrock

Async Bedrock uses aioboto3 — pass the client to AsyncSolwyn() inside the session context manager:

import os
import aioboto3
from solwyn import AsyncSolwyn

session = aioboto3.Session()
async with session.client("bedrock-runtime", region_name="us-east-1") as bedrock:
    async with AsyncSolwyn(
        bedrock,
        api_key=os.environ["SOLWYN_API_KEY"],
    ) as client:
        response = await client.converse(
            modelId="us.anthropic.claude-3-5-sonnet-20241022-v2:0",
            messages=[{"role": "user", "content": [{"text": "Hello!"}]}],
            inferenceConfig={"maxTokens": 1024},
        )

await client.converse_stream(...) returns the boto3 contract dict; iterate result["stream"] with async for and close with await result["stream"].close() or async with result["stream"]: on early abandonment. invoke_model / invoke_model_with_response_stream raise ConfigurationError exactly as on the sync client. See Amazon Bedrock.

Media surfaces

The async client serves every media surface the sync client does — the async proxies mirror the sync set one for one, with the same postures. All are awaited:

embedding = await client.embeddings.create(model="text-embedding-3-small", input="...")
image = await client.images.generate(model="gpt-image-1", prompt="...", n=1)
edited = await client.images.edit(model="gpt-image-1", image=..., prompt="...")
transcript = await client.audio.transcriptions.create(model="whisper-1", file=..., response_format="json")
speech = await client.audio.speech.create(model="tts-1", voice="alloy", input="...")
video_job = await client.videos.create(model="sora-2", prompt="...")   # OpenAI only

On a Google client, await client.models.embed_content(...), await client.models.generate_images(...) (Imagen), and await client.models.generate_videos(...) (Veo) are intercepted the same way. Each surface is budget-checked before dispatch and recorded with its modality, identically to the sync client; audio.translations warns once and passes through, and a media surface the wrapped client's adapter does not serve fails loud with UnsupportedSurfaceError. See Surface coverage for every surface's posture and the provider × modality matrix for what each provider is priced for.

Attribute passthrough

Any attribute not intercepted by AsyncSolwyn is passed through to the underlying async provider client.

Differences from Solwyn

Aspect	`Solwyn`	`AsyncSolwyn`
Context manager	`with`	`async with`
Close	`client.close()`	`await client.close()`
LLM calls	`client.chat.completions.create()`	`await client.chat.completions.create()`
Budget checks	Synchronous HTTP	Asynchronous HTTP
Reporter	Background thread queue	`asyncio.create_task`
Provider clients	`OpenAI`, `Anthropic`, `genai.Client`, boto3 `bedrock-runtime`	`AsyncOpenAI`, `AsyncAnthropic`, `genai.Client`, aioboto3 `bedrock-runtime`

The interception behavior (budget check, circuit breaker, metadata reporting) is identical. Only the I/O layer differs.

On this page