Async Streams: Production Concurrency for AI Agents

AI agent workflows are inherently concurrent. An agent generating an illustrated story needs to produce images in parallel. An ETL pipeline needs to process records in batches with backpressure. A multi-step workflow needs to respect dependency ordering while maximizing throughput. Getting concurrency right is hard. Getting it right in production -- with capacity limits, error handling, and observability -- is harder.

That is why @firefoundry/agent-sdk ships an async streaming library designed specifically for these problems. It provides three composable primitives -- PullChain, HierarchicalTaskPoolRunner, and ScheduledTaskPoolRunner -- that handle the concurrency so your agent code can focus on the business logic.

The Concurrency Problem

Consider a concrete example: the Illustrated Story tutorial. The agent receives a story prompt, generates a narrative with multiple scenes, and then produces an image for each scene. The naive approach is sequential -- generate one image at a time, wait for each to finish before starting the next. For a 10-scene story, that means 10 round trips to an image generation API, each taking 10-30 seconds. Total wall time: minutes.

The obvious fix is parallelism: fire off all 10 requests at once. But in production, that creates new problems:

• Rate limits. Most AI APIs enforce per-account or per-key request limits. Slamming 10 requests simultaneously may trigger throttling or errors.
• Resource exhaustion. If you are processing hundreds of stories concurrently, unbounded parallelism per story can overwhelm downstream services, exhaust memory, or saturate network connections.
• Multi-tenant fairness. In a shared system, one aggressive workflow should not starve other workflows of capacity.
• Dependency ordering. Some tasks depend on others. You cannot generate a summary until the individual sections are complete.

What you need is controlled concurrency -- the ability to run tasks in parallel up to a defined capacity, with backpressure, hierarchical limits, and dependency awareness. That is exactly what the async streaming library provides.

PullChain: Functional Pipelines for Async Data

PullChain is a lazy, composable pipeline for processing async data streams. If you have used functional collection methods like .map(), .filter(), and .flatMap() on arrays, PullChain will feel familiar -- except it operates on async iterables, pulling values through the pipeline on demand rather than eagerly materializing intermediate arrays.

import { PullChain, SourceBufferObj } from '@firefoundry/agent-sdk';

const result = await PullChain.from(new SourceBufferObj([1, 2, 3, 4, 5]))
    .filter(x => x % 2 === 0)
    .map(x => x * 3)
    .collect();  // [6, 12]

This looks simple, but the power is in what happens under the hood. Each stage in the pipeline is an async generator. Values are pulled through lazily -- the .map() stage does not run until .filter() yields a value, and .filter() does not pull from the source until downstream is ready for more. This means you can process arbitrarily large data streams without buffering everything in memory.

flatMap for One-to-Many Operations

Real-world pipelines often need to expand a single input into multiple outputs. A story has many scenes; a document has many pages; a batch has many records. PullChain supports .flatMap() for exactly this pattern:

const allImages = await PullChain.from(storySource)
    .flatMap(story => story.scenes)        // expand each story into its scenes
    .map(scene => generateImage(scene))    // generate an image for each scene
    .collect();

Because PullChain is pull-based, backpressure propagates naturally. If image generation is slow, the pipeline slows down rather than buffering unbounded work in memory.

Task Pool Runners: Controlled Parallelism

PullChain handles sequential pipeline processing, but what about parallel execution with capacity limits? That is the domain of the task pool runners.

HierarchicalTaskPoolRunner

The HierarchicalTaskPoolRunner solves the multi-tenant concurrency problem with two levels of capacity control: a global limit across all entities, and a per-entity limit within each entity. This is the primitive that powers parallel image generation in the Illustrated Story tutorial.

import {
  HierarchicalTaskPoolRunner,
  SourceFromIterable,
  ResourceCapacitySource
} from '@firefoundry/agent-sdk';

// Global: max 10 image generations across ALL stories
const globalCapacity = new ResourceCapacitySource({ limit: 10 });

// Per-entity: max 3 concurrent images per individual story
const perEntityCapacity = new ResourceCapacitySource({ limit: 3 });

const runner = new HierarchicalTaskPoolRunner(
  'image-generation', source, globalCapacity, perEntityCapacity
);

for await (const envelope of runner.runTasks(false)) {
  if (envelope.type === 'VALUE') results.push(envelope.value);
}

The two-level capacity model is what makes this production-ready. The global limit prevents you from overwhelming the downstream image API regardless of how many stories are being processed simultaneously. The per-entity limit ensures that a single story with 50 scenes does not monopolize all 10 global slots, starving other stories. Each story gets fair access to capacity.

Results are delivered as an async iterable of typed envelopes. Each envelope carries a type field -- 'VALUE' for successful results, or error types for failures -- so your consuming code can handle success and failure in a single iteration loop without try-catch boilerplate.

ScheduledTaskPoolRunner with DependencyGraph

Some workflows have ordering constraints. You cannot run the "load" step until "transform" completes, and "transform" cannot start until "extract" finishes. The ScheduledTaskPoolRunner combined with DependencyGraph handles exactly this pattern:

import {
  DependencyGraph,
  PriorityDependencySourceObj,
  ScheduledTaskPoolRunner
} from '@firefoundry/agent-sdk';

// Define the dependency structure
const graph = new DependencyGraph<string>();
graph.addNode('extract');
graph.addNode('transform', ['extract']);   // depends on extract
graph.addNode('load', ['transform']);       // depends on transform

// Create a priority source from the graph
const source = new PriorityDependencySourceObj(graph);

// Run with concurrency control
const runner = new ScheduledTaskPoolRunner('etl-pipeline', source, capacity);

for await (const envelope of runner.runTasks(false)) {
  // Tasks execute in dependency order, parallelizing where possible
}

The DependencyGraph is a directed acyclic graph (DAG) that tracks which nodes depend on which. When you pass it to a PriorityDependencySourceObj, the source only yields tasks whose dependencies have all completed. Tasks without mutual dependencies run in parallel up to the capacity limit. Tasks with dependencies wait until their prerequisites finish.

This is particularly powerful for multi-step agent workflows. Consider a document processing pipeline: extract text from 20 PDFs in parallel, then run entity recognition on each extracted text (again in parallel, but only after its corresponding extraction finishes), then aggregate results into a final report (only after all entity recognition steps complete). The dependency graph expresses these constraints declaratively, and the runner handles the scheduling.

Hierarchical Concurrency in Practice

The real power of these primitives emerges when you compose them. The Illustrated Story tutorial demonstrates a common pattern: entity-level parallelism within a capacity-managed pool.

Here is the mental model. You have a queue of stories to process. Each story has multiple scenes that need images. You want:

• Story-level parallelism: Process multiple stories at the same time.
• Scene-level parallelism: Within each story, generate multiple images concurrently.
• Global backpressure: Never exceed 10 total in-flight image generation requests across all stories, to respect API rate limits.
• Per-story fairness: No single story should consume more than 3 of the 10 global slots at any time.

The HierarchicalTaskPoolRunner handles all four of these constraints with a single configuration. You declare the global and per-entity capacity, feed it a source of entities and their child tasks, and iterate the results. The runner manages the scheduling, backpressure, and fairness internally.

Compare this to hand-rolling the same behavior with raw Promises or async/await. You would need a semaphore for the global limit, per-entity tracking maps, a queue for pending work, error handling for partial failures, and careful cleanup logic to release capacity on both success and failure paths. That is easily 200+ lines of brittle, hard-to-test concurrency code. The task pool runner replaces it with a declarative configuration and a for await loop.

Real-World Example: The Illustrated Story

The Illustrated Story tutorial ties these concepts together in a working agent. The workflow looks like this:

• Step 1: The agent receives a story prompt and generates a narrative with structured scene descriptions using an LLM call through the FireFoundry Broker.
• Step 2: Each scene is fed into the HierarchicalTaskPoolRunner as a child task of the story entity. The runner generates images for scenes in parallel, respecting both global and per-story capacity limits.
• Step 3: As images complete (delivered via the async envelope stream), the agent assembles the final illustrated story, matching each image to its scene.

The result is a workflow that processes stories efficiently -- maximizing throughput by parallelizing image generation -- while remaining safe for production by respecting rate limits and preventing resource exhaustion. And because results arrive as an async stream, the agent can begin assembling finished scenes while later scenes are still generating, reducing overall latency.

Note: Code examples in this article are simplified for clarity. See the Illustrated Story tutorial for a complete working example with full error handling, telemetry integration, and production configuration.

Getting Started

The async streaming primitives are included in the FireFoundry Agent SDK. If you are already building agents on FireFoundry, you have access to these tools today.

Start with PullChain for sequential pipeline processing -- it is the simplest entry point and immediately useful for any data transformation workflow. When you need parallel execution with capacity control, reach for HierarchicalTaskPoolRunner. When your workflow has dependency ordering constraints, use ScheduledTaskPoolRunner with a DependencyGraph.

All three primitives share a consistent pattern: configure a source, configure capacity, and iterate the results with for await. They compose naturally -- a PullChain can consume the output of a task pool runner, and a task pool runner can use PullChain internally for per-task processing.

For hands-on examples and the full API reference, visit the developer documentation. The Illustrated Story tutorial is the best starting point for seeing these primitives in action.

Ready to build concurrent agent workflows? If you are in the beta, start experimenting with the async streaming library today. If you are not yet in the beta, request access and we will get you set up.