Parallel AI Agent Execution in Python: Bounded Asyncio

Parallel execution helps when an agent has independent requests that can overlap. It does not make retrieved results more relevant. Set a client concurrency limit, retain each task’s success or failure, and measure completion separately from answer quality. An async function that calls blocking HTTP code still blocks its event loop.

Separate independence, capacity and quality

SearchCans Parallel Lanes describe requests in flight, not a monthly quota or a promise of a particular requests-per-second rate. Choose a client limit from the current account capacity and competing workloads. The example’s limit of two is a conservative demonstration value, not a requirement to buy a particular plan.

Use an asynchronous HTTP client inside the bound

Install httpx and use Python with asyncio support. The request callback lets a deterministic local test replace the network. The live adapter shares one AsyncClient, validates the HTTP response and parses data.organic. Each worker occupies the semaphore while it awaits its own request. The HTTPX async guide documents this client pattern; Python’s asyncio documentation explains task cancellation and timeouts.

import asyncio
import os
import httpx

def organic(envelope):
    if not isinstance(envelope, dict):
        raise ValueError("Expected JSON object")
    if envelope.get("code") != 0:
        raise ValueError("Search business error")
    data = envelope.get("data")
    if not isinstance(data, dict) or not isinstance(data.get("organic"), list):
        raise ValueError("Expected data.organic array")
    return [r for r in data["organic"] if isinstance(r, dict)
            and isinstance(r.get("link"), str) and r["link"].startswith("https://")]

async def bounded_search(queries, request, limit=2, timeout_seconds=40):
    if limit < 1 or timeout_seconds <= 0:
        raise ValueError("Positive concurrency and timeout required")
    semaphore = asyncio.Semaphore(limit)
    async def run(query):
        async with semaphore:
            try:
                envelope = await asyncio.wait_for(request(query), timeout=timeout_seconds)
                return {"query":query, "status":"ok", "results":organic(envelope)}
            except (TimeoutError, httpx.HTTPError, ValueError) as exc:
                return {"query":query, "status":"failed", "error_type":type(exc).__name__}
    return await asyncio.gather(*(run(q) for q in queries))

async def search_queries(queries, limit=2):
    async with httpx.AsyncClient(timeout=35) as client:
        async def request(query):
            response = await client.post("https://www.searchcans.com/api/v1/search",
                headers={"Authorization":"Bearer " + os.environ["SEARCHCANS_API_KEY"]},
                json={"t":"google", "s":query, "country":"us", "language":"en", "d":30000})
            response.raise_for_status()
            return response.json()
        return await bounded_search(queries, request, limit)

Verify partial success and maximum in-flight work

Use four synthetic tasks: two valid envelopes, one business error and one deliberately delayed coroutine. Instrument the callback’s active count with a try/finally decrement. The peak should not exceed two, successful task outputs should remain available, and the delayed task should produce a failed status. This tests the scheduler under controlled delays; it is not a SearchCans latency benchmark.

The timeout wraps the client operation, not work already running on the remote provider. Cancellation stops waiting locally; it does not prove that a remote job has stopped or that a retry will be free. For a large batch, replace eager task creation with a bounded queue. Limit retries by the task deadline and avoid retrying a whole batch after one failure.

Keep the scheduling result separate from the agent answer

Log task ID, query, start and finish time, status and usable source count. Do not feed a traceback into the model as source text. If an answer requires all branches, a partial result should block synthesis; otherwise make the missing branch visible to the consumer.

For result relevance use a labeled retrieval evaluation. For elapsed time and percentiles use the API timing method. the concurrency and rate-limit guide explains the terminology. This page owns bounded scheduling, rather than rewriting those three jobs into a second general agent guide.

When the bound is correct but a request pipeline still waits too long, use stage-level retrieval latency profiling to separate discovery, overlapping reads and later model work. Scheduling limits and elapsed-time diagnosis are different checks.