Async Clients

The toolkit provides both sync and async provider client initialization in doc_processing_toolkit.utils.

Functions

• initialize_clients(provider=None, logger=None, api_key=None)
  • Returns sync clients
• initialize_async_clients(provider=None, logger=None, api_key=None)
  • Returns async clients
• shutdown_clients()
  • Closes and clears every process-cached client
• await shutdown_clients_async()
  • The same, awaited on the calling event loop (see Closing clients from async code)

When provider is specified, only that provider's client is constructed and returned — the other providers' clients are never built. With provider=None, all available clients are constructed and returned as a dict. Construction (not just the return value) is gated on provider so unrequested SDK clients are never created and orphaned to garbage collection, which would cost file descriptors and could emit teardown noise (e.g. the Fireworks SDK eagerly opens aiohttp sessions per client when an event loop is running).

Both call forms are served from, and published to, the shared cache. The all-providers form used to bypass it, rebuilding one client per provider on every call and orphaning the previous set — the multiplication behind a file-descriptor exhaustion incident. An explicit api_key still bypasses the cache entirely, for credential isolation.

api_key overrides the provider's environment variable for that one call. It is only honoured together with an explicit provider, and clients built from it bypass the shared cache described below.

Client caching and ownership

Provider clients are cached for the lifetime of the process, so repeatedly constructing Agents for the same provider reuses one SDK client instead of opening a new connection pool per agent (the failure mode this prevents is file-descriptor exhaustion — OSError: [Errno 24] Too many open files).

The cache key is (provider, "sync" | "async"), with two rules layered on top:

• Async entries are event-loop-affine. An async client captures the loop that created it, so a cached entry is only handed back to a caller running on that same loop. Anything else — including a second asyncio.run(...) in the same process — is treated as a cache miss and gets a freshly built client. This is what keeps run_sync() from ever surfacing RuntimeError: Event loop is closed.
• Explicit keys are never shared. A client built with api_key=... bypasses the cache entirely in both directions: it is not served from the cache and it is not stored into it. Credentials from one caller can never leak into another caller's client.

What Agent.close() closes

close() follows the ownership rule implied by the cache:

How the Agent got its client	Owned by the Agent?	Closed by await agent.close()?
Default (client came from the process cache)	No	No — closing it would break every other live Agent on that provider
Constructed with Agent(api_key=...)	Yes	Yes

close() always disconnects local MCP servers and drops the Agent's client references, regardless of ownership.

For a clean process exit — an embedder shutting down, a worker draining, a test suite finishing — call shutdown_clients() to close and clear the shared cache:

from doc_processing_toolkit.utils import shutdown_clients

shutdown_clients()

Sync clients are closed directly. Async clients are closed on the event loop that created them while that loop is still alive — from inside that loop the close completes after the call returns; from another thread it is waited for briefly. A client whose loop has already been closed is dropped for garbage collection instead (its transports cannot be driven from another loop, and collection closes its sockets). shutdown_clients() never raises.

Closing clients from async code

shutdown_clients() is synchronous. Called from a plain thread it waits for the async clients' connection pools to close, but called from a thread that is already running an event loop — an ASGI shutdown handler, a Jupyter cell, a FastMCP tool handler — it can only schedule those coroutine closes and returns before they complete. Await shutdown_clients_async() there instead:

from contextlib import asynccontextmanager

from fastapi import FastAPI

from doc_processing_toolkit.utils import shutdown_clients_async


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await shutdown_clients_async()


app = FastAPI(lifespan=lifespan)

It closes the async clients bound to the calling loop under a five-second budget, closes cached sync clients inline, and discards — without awaiting — any async client belonging to a different or already-closed loop, because only that loop could drive its transport. It is idempotent, safe when nothing is cached, and never raises. Clients owned by an Agent(api_key=...) are not in the shared cache and are closed by Agent.close() instead.

Like shutdown_clients, it is not re-exported at the package root; import it from doc_processing_toolkit.utils.

You rarely need to close anything by hand

An event loop that owns cached async clients now closes them on its way out, before the loop itself closes — so a sequence of run_sync(...) calls, each of which runs on its own throwaway asyncio.run loop, no longer leaves one orphaned transport behind per call. Five sequential synchronous runs used to orphan four. This is what removed the Unclosed client session / Unclosed connector warnings the Fireworks SDK's aiohttp transport emitted; on every provider it stops the file-descriptor drip. The same hook fires when a long-lived server loop shuts down.

Call shutdown_clients_async() (or shutdown_clients()) explicitly when you want the pools released at a specific moment rather than at loop teardown. Cache-served synchronous clients are unaffected and stay open for the life of the process, as before.

Agent.cleanup() was removed

Agent previously carried a no-op cleanup() plus an __aenter__/__aexit__ pair that shadowed the real async-context-manager behaviour. All three were deleted. async with Agent(...) still works and now reaches close() as intended; explicit await agent.close() is unchanged.

.env is read once per process

.env is loaded on the first client initialization in the process and never re-read. Editing .env, or rotating a key in it, therefore has no effect on an already-running process — restart to pick up new credentials. This is deliberate: re-reading .env on every initialization cost a file open per call and made credential state depend on wall-clock timing. To vary credentials within one process, pass api_key explicitly instead.

Initialization failures name their real cause

When a client cannot be built, the raised RuntimeError distinguishes two situations:

• The key is genuinely unset — the message names the exact environment variable, e.g. anthropic API key not configured (ANTHROPIC_API_KEY is unset).
• Anything else — the underlying SDK construction error is chained onto the RuntimeError (visible as __cause__ and in the traceback), so an OSError: [Errno 24] Too many open files reads as itself rather than as a phantom credential problem.

doc_processing_toolkit.utils.get_client_init_error(provider, kind) returns the last recorded construction failure for a (provider, "sync" | "async") pair, or None if the last attempt succeeded. PROVIDER_API_KEY_ENV maps each provider name to its key environment variable.

Related: load_model_information() reads the bundled model_information.json once per process and raises RuntimeError (chained from the original OSError/ValueError) when the file cannot be read or parsed. It never returns None, so a packaging or file-descriptor problem fails at the call site instead of decaying into a downstream 'NoneType' object is not subscriptable.

How Agent uses them

• Function-calling path in Agent uses async client accessors.
• Structured-output path uses sync client accessors.
• run_sync(...) is a wrapper over async run(...) with event loop handling.

When to use each

• Use async (await agent.run(...)) in async apps, services, and test frameworks.
• Use run_sync(...) in scripts and synchronous contexts.
• Prefer one concurrency style per boundary to reduce loop complexity.

Practical caveats

• run_sync(...) is convenience-first and handles common loop scenarios, but async-first application code should still prefer await agent.run(...).
• Mixed sync/async provider code paths may expose different timing/logging behavior; rely on returned AgentResult fields instead of transport-specific assumptions.
• Do not carry your own async client across event loops. This matters only if you build an AsyncOpenAI (or any async SDK client) yourself and pass it to a provider function — Agent and initialize_async_clients(provider=...) already hand out one client per loop. A client's connection pool belongs to the loop that first used it. Under openai 3.x (httpx2) the first request made on a different loop fails with RuntimeError: Event loop is closed; openai 2.x (httpx) silently tolerated the same misuse, so code that reused one client across asyncio.run(...) calls can start failing after the upgrade. Create the client inside the loop that uses it.

Token Counting

An async token counting function is available for OpenAI models:

• async_count_tokens_openai(text, model) — performs async token counting using tiktoken, useful in async pipelines where the synchronous count_tokens() would block.
• async_count_tokens_openai_with_source(...) — the same count plus a token_source label naming how it was obtained. See Cost Tracking.

Client Timeouts

All provider clients are now initialized with generous timeouts to accommodate long-running model generations:

Timeout	Value
Read	2 hours
Write	10 minutes
Connect	30 seconds

These timeouts work alongside streaming transport (where each chunk resets the idle timer) to virtually eliminate timeout-related failures.

SDK-era client kwargs

Fireworks clients are built from constructor kwargs that depend on which fireworks-ai era is installed. On >= 1.x the client is given max_retries=0, because that SDK retries twice by default and those retries would stack multiplicatively with the toolkit's own retry policies; the legacy (< 1.x) constructor has no max_retries parameter and must not receive one. Both sync and async initialization share the same gating, so initialize_clients and initialize_async_clients behave identically here. See Providers.

Related recipe

• Sync Wrappers