Container-enabled asyncio is all you need (to build Pythonic AI workflows at scale)

You built an agentic workflow. It calls three LLM providers in parallel, fans out a hundred retrieval tasks, runs ten tools that make API calls, streams responses, and returns a final answer. On your laptop with 10 inputs, it works beautifully.

Then you run it in production with 10,000 inputs against a real rate limit, and it falls over. Connections pile up. One slow connection stalls the whole batch. An OOM in a single tool call kills the entire process. And you can't back-pressure inbound work without rewriting half of it.

So you reach for a framework: a workflow DSL, a YAML DAG, a graph library. The instinct behind that reach is usually some version of "Python isn't enough to express this, we need an orchestrator with a specialized DSL." This is the talk I gave at PyCon US 2026, and I want to push back on that instinct. The same story plays out if you swap "agentic workflow" for a classical ML training sweep or an ETL pipeline: heterogeneous compute, one step that OOMs, one slow node that stalls the batch. The surface differs, but the overall shape doesn't.

I recently talked about this problem at Pycon 2026, which you can watch here:

Production AI is a graph of waits

Here's the reframe. Production AI workflows are dynamic execution graphs of waits, where:

1. The shape of the graph depends on inputs or intermediary outputs.
2. Workloads vary in concurrency, parallelism, and backpressure.
3. Nodes are composed of variable compute units (CPU, GPU, memory).

The first two requirements are language-level problems, and `asyncio` is the standard library's answer to both. Your bottleneck is rarely "one model call." Production paths are graphs of overlapping waits: HTTP round-trips, disk reads, queue puts, GPUs queued ahead of you in the cluster. Latency is the sum of overlapped (or serialized) waits, not how fast your `for` loop runs.

The third requirement is a systems-level problem, and no DSL solves it. Heterogeneous compute, OOM containment, isolation of compute and permissions: those are the job of a container orchestrator that plays well with Python.

So there are two responsibilities, and they belong in two different layers:

• Coordinating waits is a language-level problem. `asyncio` solves this.
• Isolating compute is a systems-level problem. Containers solve this.

When you keep those two layers clean, you don't need a third grammar in the middle. The awkward middle ground a lot of teams land in is adopting another grammar (a DSL, a YAML DAG) before reaching for stdlib `asyncio`, which already fits I/O-heavy coordination and keeps your logic in debuggable Python. The cost of jumping early is real. It means you end up with control flow you can no longer step through, retries you can no longer reason about, and debugging that means reading framework internals instead of your own code.

asyncio already has the primitives you wanted from a DSL

`asyncio` is a scheduler in one thread. It holds two collections, ready tasks and tasks waiting on I/O, and while one task awaits readiness the loop runs the others. That cooperative scheduling is the whole trick behind concurrency: one cook weaving between boiling water, chopping garlic, and simmering sauce gets dinner out faster than doing each step start-to-finish.

The reason I push back on the DSL instinct is that most of what a workflow DSL sells you is already sitting in the standard library. Here's the mapping.

Fan-out with `asyncio.gather`. Run coroutines at the same time, and keep the completed results even when some fail:

async def parallel_llm_calls(payloads: list[dict]) -> list[dict]:
    return await asyncio.gather(
        *(client.post("/v1/chat", json=p) for p in payloads),
        return_exceptions=True,  # partial failure without losing the rest
    )

Structured concurrency with `asyncio.TaskGroup` (3.11+). Scope child tasks so the block doesn't proceed until every task inside it finishes, which is exactly what you want before a downstream embedding or processing step:

async def fetch_all(urls: list[str]) -> list[str]:
    async with asyncio.TaskGroup() as tg:
        handles = [tg.create_task(download_contents(u)) for u in urls]
    return [t.result() for t in handles]

Bounded concurrency with `asyncio.Semaphore`. Cap the number of in-flight calls so you don't trip a server-side rate limit:

sem = asyncio.Semaphore(8)  # at most 8 concurrent calls

async def _call_llm(payload: dict):
    async with sem:
        return await fetch(payload)

Backpressure with `asyncio.Queue(maxsize=...)`. A bounded queue makes producers block when it's full, which is the entire point: the queue pushes back on inbound work instead of letting connections pile up:

q: asyncio.Queue[Doc] = asyncio.Queue(maxsize=200)

async def producer(docs):
    for d in docs:
        await q.put(d)   # blocks when full -> backpressure
    await q.join()       # wait until drained

async def worker():
    while True:
        d = await q.get()
        await train_model(d)
        q.task_done()

Step timeouts with `asyncio.timeout` (3.11+). Bound a stalled call so you can retry on your terms instead of the provider's:

async def call_tool(name: str, args: dict, attempts: int = 3):
    for attempt in range(attempts):
        try:
            async with asyncio.timeout(30):
                return await api_provider.invoke(name, args)
        except (TimeoutError, ProviderError):
            if attempt == attempts - 1:
                raise
            await asyncio.sleep(2 ** attempt)  # plain backoff

Streaming with `async for`. Send tokens to a websocket as they arrive, at constant memory instead of buffering the full completion:

async with client.stream("POST", "/v1/chat", json=payload) as resp:
    async for chunk in resp.aiter_text():
        await websocket.send_text(chunk)

Put those next to the feature list of the DSL you were about to adopt:

And you still get `if/elif/else`, `try/except`, and `for/while` for free, because it's just Python. No new grammar. `asyncio` can look intimidating, but I'd encourage you to actually learn it. It's pretty fun.

What asyncio can't do, and what containers can

Concurrency is necessary but not sufficient for production AI at scale. `asyncio` cannot, by itself:

• Provision a GPU for a training step.
• Give you OOM containment when a tool call blows past its memory.
• Isolate a noisy tenant from the rest of your platform.
• Scale horizontally without you rewriting your mental model.

Those are scheduler-level concerns, not language-level concerns. Containers on something like Kubernetes give you the hard boundaries `asyncio` explicitly delegates: GPU and CPU pools, memory quotas, OOM containment so one bad batch doesn't take down the hub, tenancy and blast-radius limits per workload, and job-level retries and deadlines. The thing they don't change is your user code. You don't have to conform your logic to a different concurrency paradigm to get isolation.

The pattern that ties the two layers together is this: when you need isolation, `asyncio` doesn't disappear, it becomes the client. Your `async def` service talks to queues, provider APIs, and the cluster API. Pods hold the heavy or risky pieces (training, sandboxed tool execution, anything GPU- or OOM-bound). Where a piece of work runs becomes runtime configuration, not a second concurrency model living in your source. One line for the room: the event loop coordinates, the container orchestrator isolates.

A reference implementation you can read: Flyte 2

I don't want to leave this as an assertion, so here's an open-source example of the asyncio-as-client pattern you can actually inspect. Flyte 2 is an Apache 2.0 project owned by the Linux Foundation. The point isn't the product, it's that the design is legible: it's `async def` all the way down, with a control plane doing the isolation.

You start from plain async Python and add compute configuration around it. The functions don't change:

import asyncio
import flyte

env = flyte.TaskEnvironment(
    name="hello_world",
    image=flyte.Image.from_debian_base().with_pip_packages("numpy"),
    resources=flyte.Resources(cpu=2, memory="1Gi"),
)

@env.task(retries=3, cache="auto")   # retries + caching, no rewrite
async def predict(x: int) -> int:
    return 2 * x + 5

@env.task
async def main(data: list[int]) -> float:
    xs = await asyncio.gather(*(predict(x) for x in data))
    return sum(xs) / len(xs)
    

if __name__ == "__main__":
    flyte.init_from_config()
    flyte.run(main, data=range(10))

Run it as a normal Python process with `asyncio.run(main(...))`, or point it at a cluster with `flyte.run(main, data=range(10))` and each `Run` and `Action` gets its own pod for isolation and reproducibility. The programming model stays `async def`; deciding where work runs is a one-line change.

The part worth pausing on is what's underneath. Flyte's controller is not a bespoke scheduler with its own grammar, it's the exact `asyncio` primitives from earlier, composed. A `TaskGroup` of worker coroutines drains a bounded `asyncio.Queue` of actions, and transient remote errors back off and re-enqueue the same action:

class Controller:
    def __init__(self, workers: int = 20, ...):
        self._shared_queue: asyncio.Queue[Action] = asyncio.Queue(maxsize=10000)
        self._workers = workers

    async def _bg_worker_pool(self):
        async with asyncio.TaskGroup() as tg:
            for i in range(self._workers):
                tg.create_task(self._bg_run(f"worker-{i}"))

    async def _bg_run(self, worker_id: str):
        while self._running:
            action = await self._shared_queue.get()
            try:
                await self._bg_process(action)
            except flyte.errors.SlowDownError:
                action.retries += 1
                await asyncio.sleep(self._backoff(action.retries))
                await self._shared_queue.put(action)  # re-enqueue
            finally:
                self._shared_queue.task_done()

The remote controller adds two more of the same primitives: a per-parent `asyncio.Semaphore` so one task can't starve its siblings, and an `AsyncLimiter` so every worker respects a global QPS cap when talking to the control plane:

class RemoteController(Controller):
    def __init__(self, ..., default_parent_concurrency=1000, max_qps=100):
        super().__init__(...)
        self._parent_action_semaphore = defaultdict(
            lambda: asyncio.Semaphore(default_parent_concurrency)
        )
        self._rate_limiter = AsyncLimiter(max_qps, 1.0)  # global QPS cap

    async def submit(self, _task, *args, **kwargs):
        parent = unique_action_name(current_action_id)
        async with self._parent_action_semaphore[parent]:   # fair-share
            return await self._submit(_task, *args, **kwargs)

    async def _bg_launch(self, action: Action):
        async with self._rate_limiter:                      # respect limits
            await self._actions_service.enqueue(...)        # gRPC to control plane

If you learned the six primitives in the section above, you can read this. That's the whole argument in one code sample: the orchestrator isn't a foreign grammar you have to trust, it's `Queue`, `TaskGroup`, `Semaphore`, and a rate limiter wired to a gRPC control plane.

Crossing a container boundary isn't free, and it's worth being honest about the trade you're making. Inside one process you share Python objects, function calls are cheap, and there's a single failure domain. Across isolated workers you serialize arguments through files and object stores, you pay network and serialization per step, and you inherit partial failure and retries at the workflow scope. `asyncio` helps you express fan-out and failure policy in code. It doesn't erase distributed semantics.

Tomorrow, before you reach for a DAG tool

Two things to take with you.

• Learn a little more `asyncio`. `TaskGroup`, `gather`, `Semaphore`, `Queue`, `timeout`, and `Condition` cover most of what a DSL sells you, and code that stays in plain Python stays reviewable by humans and by agents.
• Don't confuse "I need isolation" with "I need a DSL." Provisioning compute, OOM containment, multi-tenancy, and permission isolation are a container orchestrator's job, not a new grammar's. Keep the coordination layer and the isolation layer clean and you won't need a third thing in the middle.

Nearly every library you already lean on for AI work (FastAPI, Pydantic AI, LiteLLM, vLLM, HTTPX, LangGraph, Ray, Flyte) is `asyncio` underneath. If you want to fully use, debug, or understand any of them, learning `asyncio` pays for itself. The best DSL for AI was in front of you the whole time.

Try it

uv pip install flyte

# start a local Flyte devbox
flyte start devbox

# create config
flyte create config \
    --endpoint localhost:30080 \
    --project flytesnacks \
    --domain development \
    --builder local \
    --insecure

# run the async example above
uv run flyte_hello_world.py

The talk is on YouTube, the abstract is on the PyCon US 2026 schedule, and Flyte's queue, run, and state services are documented alongside the protos in the Flyte backend.