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,
    tags: Mapping[str, str] | None = None,
    on_unmetered: str | None = None,          # "warn" | "raise" | "allow"
    acknowledge_untracked: Collection[str] | None = None,
    selection_policy: SelectionPolicy | None = None,
    control_plane_transport: ControlPlaneTransport | None = None,
    **config_kwargs,
)

Parameters

Everything after client is keyword-only. 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

The same set as Solwyn: ConfigurationError at construction (invalid API key, malformed fallback, unknown config field, provider= / client-family mismatch, invalid tags, invalid acknowledgment token, non-finite failover timeout) and at call time (the Bedrock invoke_model family, and Responses background / streaming parse / metering-critical extra_body); UntranslatableRequestError and UntranslatableModelError; UntrackedSpendSurfaceError under on_unmetered="raise"; UnsupportedSurfaceError; and RunStoppedError inside a stopped run.

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.

It then forwards to the wrapped provider client's own close — aclose() if the client has one, otherwise close() — and awaits the result when it is awaitable.

Changed in 0.4.0. Two things:

• close() is bounded by one wall-clock deadline (reporter_shutdown_deadline, default 5.0 seconds), set on the constructor rather than passed here. Undelivered work at the deadline is counted and dropped — see Undeliverable spend. Held budget leases are surrendered on the way out.
• The reporter's flush loop now auto-starts on the first enqueue, so an AsyncSolwyn built without async with no longer queues events and settlements silently until close(). async with is still the recommended pattern. See Async reporters start on first use.

close() is the only method AsyncSolwyn adds. The run APIs — solwyn.run(...), solwyn.start_run(...), solwyn.create_run(...) and its RunHandle, solwyn.current_run_context() — are module-level functions on the solwyn package and work the same under async with; their signatures are in the API summary on Agent runs.

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!"}],
)

# SDK v0.6.0+: the Responses API is metered too
response = await client.responses.create(model="gpt-4o", input="Hello!")
async with client.responses.stream(model="gpt-4o", input="Hello!") as stream:
    async for event in stream:
        ...

responses.create, responses.parse, and the responses.stream() helper are metered on native OpenAI and Azure OpenAI, with the same refusals as the sync client. The async stream() helper is returned without awaiting and defers its budget check to the first __aenter__; a helper that is never entered performs no check. See The Responses API.

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 is untracked and follows on_unmetered, 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 and type transparency

Any attribute not intercepted by AsyncSolwyn is resolved against the underlying async provider client through the coverage guard: untracked leaves follow on_unmetered, namespaces come back guarded, and structural attributes are silent. Writes and deletes forward to the provider client.

SDK v0.6.0+: isinstance(client, AsyncOpenAI) is True and client.__class__ is AsyncOpenAI, while type(client) stays AsyncSolwyn. copy.copy / copy.deepcopy return the same wrapper and pickle raises TypeError. Details are identical to the sync client — see Type transparency.

Differences from Solwyn

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