Bound concurrent calls and validate results

Learning objective

Apply Python async/await, asyncio.gather, timeouts, structured validation (Pydantic and JSON Schema), and exception handling.

Core idea

For an engineer coordinating agent dependencies in Python, concurrency, deadlines, and validation solve different problems. async and await organize cooperative asynchronous work. asyncio.gather() runs independent awaitables concurrently. A timeout bounds waiting, and a validator checks whether a returned value satisfies the application’s structure contract.

None of those proves business success. A request can return before its deadline yet contain invalid data. A valid object can describe a denied operation. A caller timeout can leave a remote action’s outcome uncertain. Keep each outcome explicit rather than turning every failure into an empty result or an automatic retry.

How it works

Await work without blocking the event loop

Calling an async def function creates a coroutine object; it does not by itself run that coroutine to completion. Await it or schedule it through an appropriate asyncio mechanism. asyncio.run() provides an entry point for a standalone program, while code already running in an event loop should await its operation within that loop.

Awaiting an asynchronous network operation lets other scheduled tasks make progress while the operation waits. Sequential awaits still impose a sequence: if you await source A and only then start source B, B cannot overlap A’s wait. Use concurrency when the operations are independent, not when B requires A’s validated output.

Asyncio tasks use cooperative scheduling. Declaring a function async does not make blocking work nonblocking or make CPU work parallel. For an I/O-bound blocking function that must run in this design, Python documents asyncio.to_thread() as a way to execute it without blocking the event loop. Prefer an asynchronous client when the dependency provides one.

Choose gather’s failure semantics deliberately

await asyncio.gather(a(), b()) schedules coroutine arguments as tasks and returns results in input order, not completion order. That ordering lets you associate each result with its source even when B finishes first. Concurrent completion does not erase the need for stable source identifiers.

With the default return_exceptions=False, the first raised exception propagates to the caller. Other submitted awaitables are not automatically cancelled and can continue running. Catching the first failure therefore does not mean sibling work has stopped. Decide whether the application needs all individual outcomes or a group that terminates when a member fails.

With return_exceptions=True, gather includes exceptions in the ordered result list. Inspect each entry before interpreting it as data. This supports an application that records success or failure independently for every source. It does not turn an exception into a valid empty answer. If the requirement is structured group failure, asyncio.TaskGroup provides stronger sibling-cancellation guarantees than gather; select it according to that requirement.

Put a deadline around the intended unit

Python 3.11 and later provide asyncio.timeout(). Its argument is a delay in seconds. Place the context around the complete unit whose waiting you intend to bound, such as a group of concurrent source calls. Independent per-call timeouts do not necessarily define the overall response budget when calls are sequential or followed by retries.

The timeout context cancels the current task and converts the resulting cancellation into TimeoutError. Catch TimeoutError outside the async with block. If cancellation reaches gather, unfinished submitted awaitables are cancelled. Cancellation is cooperative, however; do not interpret a local timeout as proof that a downstream service rolled back a write.

Use try/finally for cleanup that must run on both success and cancellation. asyncio.CancelledError directly subclasses BaseException, and Python recommends propagating it after cleanup when explicitly caught. Swallowing cancellation can break timeout and structured-concurrency behavior. A timeout is also not a retry policy: determine eligibility and remaining budget separately.

Validate structure at the receiving interface

Pydantic’s BaseModel expresses a Python data contract. model_validate_json() parses and validates JSON against that model and raises ValidationError when validation fails. model_json_schema() exports the model’s JSON Schema. Schema publication and runtime validation are separate actions; handing a schema to a producer does not validate its response in your process.

For Azure OpenAI’s Responses API with OpenAI Python 2.x, responses.parse(text_format=…) accepts a Pydantic model, converts it to JSON Schema under text.format, and exposes the parsed value in output_parsed; see 0.3 for constraining model output. Results from non-model dependencies, such as tools, A2A artifacts, and REST services, still need validation where you receive them.

Pydantic can coerce input types by default. Choose strict validation when coercion would hide a contract violation; model_validate_json(..., strict=True) enables strict behavior for that call. Strict JSON validation still has type-specific rules, so test the types your contract uses instead of assuming a universal ban on every conversion. ConfigDict(extra="forbid") rejects unexpected fields; the default extra-field behavior is to ignore them.

JSON Schema describes the interchange contract independently of Python classes. For object schemas, properties describes named fields, but does not make them mandatory. Use required for mandatory fields, and additionalProperties: false to reject undeclared fields in a simple object schema. A schema must be applied by a validator to check an instance; it is not an execution engine or an authorization policy.

The following runnable example uses synthetic asynchronous sources to show collection and validation. It requires Python 3.11 or later and Pydantic 2:

import asyncio
from pydantic import BaseModel, ConfigDict, ValidationError

class Evidence(BaseModel):
    model_config = ConfigDict(extra="forbid")
    source_id: str
    passage: str

async def read_source(source_id):
    await asyncio.sleep(0)
    return f'{{"source_id":"{source_id}","passage":"Example evidence"}}'

async def collect():
    async with asyncio.timeout(2):
        outcomes = await asyncio.gather(
            read_source("A"), read_source("B"), return_exceptions=True
        )
    records = []
    for outcome in outcomes:
        if isinstance(outcome, BaseException):
            raise outcome
        records.append(Evidence.model_validate_json(outcome, strict=True))
    return records

try:
    print([item.source_id for item in asyncio.run(collect())])
except TimeoutError:
    print("Collection timed out")
except ValidationError:
    print("Source violated the evidence contract")

The output is ['A', 'B']. The two-second value is an illustrative application budget, not a service limit. The example validates after collection, so that timeout bounds source waiting rather than synchronous validation. Production code must account for validation time in its overall budget and replace the synthetic sources with its dependency adapters. The loop rejects the bundle on an exception or invalid record; partial acceptance requires an explicit per-source outcome contract. Inside one deadline, gather overlaps both sources; after it, any exception or invalid record rejects the bundle, and expired waiting surfaces as TimeoutError outside the block.

Key distinctions and trade-offs

Concurrency versus dependency: Gather overlaps independent work. A dependent operation should start only after its prerequisite succeeds and validates.

Timeout versus cancellation versus remote outcome: A timeout reports expired waiting; cancellation requests local interruption. Neither establishes whether an external side effect committed.

JSON versus schema validity versus business validity: Parsing accepts JSON syntax. Schema validation checks structure and selected constraints. Business validation checks meaning, evidence, entitlement, and allowed action; see 0.6 and 0.8.

Strict validation exposes producer defects but can reject inputs a coercing interface accepts. Partial results improve availability only when the product defines which sources are optional. An empty fallback value that hides an authorization failure or a malformed required result changes the contract.

Architecture / decision scenario

An assistant obtains inventory and pricing independently, then prepares a proposal requiring both. The orchestrator starts both asynchronous calls with gather and a shared timeout. It deliberately uses return_exceptions=True rather than fail-fast sibling cancellation so it can record both sources’ outcomes and identify which source failed, even though both must succeed.

After collection, it validates each JSON response with the appropriate Pydantic model, strict settings, and forbidden extra fields. The published JSON Schema marks required identifiers explicitly. A successful transport response containing an unexpected price type produces ValidationError, not a valid price or an empty quote. If either required source fails, the application refuses to prepare the proposal and records which source failed.

If the deadline expires, the caller handles TimeoutError outside the timeout context and performs cleanup without swallowing cancellation. It does not claim that a timed-out downstream reservation was undone. Authorization errors, dependency failures, validation failures, and expired waiting remain different outcomes with different recovery decisions.

Common exam traps

• Assuming calling an async function runs its body to completion.
• Expecting sequential awaits to run concurrently.
• Assuming default gather cancels siblings after the first exception.
• Using return_exceptions=True without checking the returned entries.
• Catching the timeout context’s TimeoutError inside that context.
• Assuming Pydantic rejects coercion or extra fields by default.
• Treating JSON Schema properties as a required-field declaration.
• Equating a local timeout or structurally valid response with business success.

What to remember

Schedule independent work deliberately, select gather’s failure behavior, and bound waiting at the intended level. Validate returned structures at the receiving interface, then apply business checks. Preserve timeout, cancellation, dependency failure, and validation failure as distinct outcomes.

Knowledge check

References

• Python Software Foundation, Coroutines and tasks
• Pydantic, Models
• Pydantic, Strict mode
• JSON Schema, Object
• Microsoft, Structured outputs