Scaling Agent Systems Basics

8 examples to get you started with scaling agent work - 5 basic and 3 intermediate.

You will build a tiny job queue, process agent jobs with a worker, cap concurrency, sketch backpressure, and separate intake from the slow model/tool loop.

Prerequisites

Python 3.11+ recommended
Comfort with threads or asyncio-style concurrency concepts
No paid APIs required for these sketches (fake model and tools)

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Run an Agent Job Inline (The Non-Scalable Default)

Start from the pattern that fails under load: do everything inside the request path.

import time
 
 
def fake_model(prompt: str) -> str:
    time.sleep(0.05)  # stand-in for network + generation
    return f"answer:{prompt[:20]}"
 
 
def run_agent_inline(user_goal: str) -> str:
    # In real agents this is many model/tool turns
    return fake_model(user_goal)
 
 
print(run_agent_inline("summarize ticket 42"))

Inline runs couple HTTP latency to every model and tool turn.
Under concurrent users, each request opens more provider load with no shared budget.
Scaling starts by decoupling intake from execution.

Related: What Actually Bottlenecks an Agent System at Scale

2. Put Agent Jobs on a Queue

Enqueue work, then process it separately.

from dataclasses import dataclass, field
from queue import Queue
from typing import Any
import uuid
 
 
@dataclass
class AgentJob:
    job_id: str
    goal: str
    meta: dict[str, Any] = field(default_factory=dict)
 
 
jobs: Queue[AgentJob | None] = Queue()
 
 
def enqueue(goal: str) -> str:
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    jobs.put(job)
    return job.job_id
 
 
job_id = enqueue("summarize ticket 42")
print("enqueued", job_id, "depth", jobs.qsize())

The API (or CLI) only accepts work and returns a job id.
A worker owns the slow loop.
Depth is your first capacity signal.

3. Process Jobs With One Sequential Worker

Drain the queue one job at a time.

def process_job(job: AgentJob) -> dict:
    result = fake_model(job.goal)
    return {"job_id": job.job_id, "result": result, "status": "done"}
 
 
def worker_loop(q: Queue[AgentJob | None]) -> None:
    while True:
        job = q.get()
        if job is None:
            q.task_done()
            break
        print(process_job(job))
        q.task_done()
 
 
enqueue("goal-a")
enqueue("goal-b")
jobs.put(None)  # poison pill to stop demo
worker_loop(jobs)

Sequential processing is a natural concurrency limit of 1.
Many production agent fleets start here and only raise concurrency when metrics allow.
Prefer durable brokers later; the shape stays the same.

4. Store Results Outside the Worker Memory

Clients need a place to poll status.

from threading import Lock
 
results: dict[str, dict] = {}
_lock = Lock()
 
 
def process_and_store(job: AgentJob) -> None:
    out = process_job(job)
    with _lock:
        results[job.job_id] = out
 
 
def get_status(job_id: str) -> dict:
    with _lock:
        return results.get(job_id, {"job_id": job_id, "status": "pending"})
 
 
j1 = enqueue("status demo")
# In a real system the worker runs in another process
process_and_store(jobs.get())
print(get_status(j1))

Never require the original HTTP connection to stay open for multi-minute agents.
Persist results (Redis, Postgres, object store) so restarts do not lose outcomes.
Return pending / running / done / failed with a stop reason when applicable.

5. Cap How Many Jobs Start at Once

Raise throughput carefully with a semaphore.

from concurrent.futures import ThreadPoolExecutor
from threading import Semaphore
 
MAX_IN_FLIGHT = 2
sem = Semaphore(MAX_IN_FLIGHT)
 
 
def run_with_cap(job: AgentJob) -> dict:
    with sem:
        return process_job(job)
 
 
def drain_with_pool(pending: list[AgentJob]) -> list[dict]:
    with ThreadPoolExecutor(max_workers=4) as pool:
        # max_workers can be higher than MAX_IN_FLIGHT; sem is the real gate
        return list(pool.map(run_with_cap, pending))
 
 
batch = [AgentJob(str(i), f"goal-{i}") for i in range(5)]
print(drain_with_pool(batch))

Pool size is not the same as allowed concurrent agent runs.
Cap against provider RPM/TPM and tool bulkheads, not only CPU.
Full recipe: Concurrency Limits and Backpressure for Agent Workers.

Intermediate Examples

6. Reject New Work When the Queue Is Too Deep

Backpressure starts at admission.

MAX_DEPTH = 3
 
 
def enqueue_with_limit(goal: str, q: Queue[AgentJob], max_depth: int = MAX_DEPTH) -> str:
    if q.qsize() >= max_depth:
        raise RuntimeError("queue_full: try later or shed load")
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    q.put(job)
    return job.job_id
 
 
q2: Queue[AgentJob] = Queue()
for g in ["a", "b", "c", "d"]:
    try:
        print("ok", enqueue_with_limit(g, q2))
    except RuntimeError as e:
        print("rejected", e)

Unbounded queues only hide overload until latency explodes.
Map queue_full to HTTP 429/503 with Retry-After in real APIs.
Pair with cheaper-model degrade under pressure when the product allows it.

7. Prefer Fake Cache Hits Before Spending a Model Call

Caching is free capacity.

import hashlib
 
cache: dict[str, str] = {}
 
 
def cache_key(goal: str) -> str:
    return hashlib.sha256(goal.strip().lower().encode()).hexdigest()
 
 
def run_cached(goal: str) -> tuple[str, bool]:
    key = cache_key(goal)
    if key in cache:
        return cache[key], True
    value = fake_model(goal)
    cache[key] = value
    return value, False
 
 
print(run_cached("FAQ: reset password"))
print(run_cached("FAQ: reset password"))  # hit

Cache only deterministic, safe, non-personalized results unless you key by tenant and auth scope.
Tool-result caches often save more money than final-answer caches.
Expand in Caching Agent Responses and Tool Results to Cut Redundant Calls.

8. Sketch Cost Pressure Routing

When spend pressure is high, choose a cheaper model path.

from dataclasses import dataclass
 
 
@dataclass
class Budget:
    spent: float = 0.0
    soft_limit: float = 1.0
 
 
def pick_model(budget: Budget) -> str:
    if budget.spent >= budget.soft_limit * 0.8:
        return "cheap-small"
    return "premium-large"
 
 
def run_with_routing(goal: str, budget: Budget) -> dict:
    model = pick_model(budget)
    # pretend costs
    cost = 0.02 if model == "cheap-small" else 0.15
    budget.spent += cost
    return {"model": model, "result": fake_model(f"{model}:{goal}"), "spent": budget.spent}
 
 
b = Budget(spent=0.7, soft_limit=1.0)
print(run_with_routing("hard task", b))
print(run_with_routing("easy task", b))

Cost-aware scale is a scaling strategy, not only a billing feature.
Always eval quality when you demote models under load.
Deep dive: Cost-Aware Scaling: Routing Load to Cheaper Models Under Pressure.

What You Learned

Decouple intake from the agent loop with a queue and job ids.
Process with workers; store results for polling or push notify.
Cap in-flight runs separately from thread/process pool size.
Apply admission control when depth is high; cache and cheaper routes free capacity.

Next Steps

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Scaling Agent Systems Basics

8 examples to get you started with scaling agent work - 5 basic and 3 intermediate.

You will build a tiny job queue, process agent jobs with a worker, cap concurrency, sketch backpressure, and separate intake from the slow model/tool loop.

Prerequisites

Python 3.11+ recommended
Comfort with threads or asyncio-style concurrency concepts
No paid APIs required for these sketches (fake model and tools)

python -m venv .venv && source .venv/bin/activate
# stdlib only for the examples below

Basic Examples

1. Run an Agent Job Inline (The Non-Scalable Default)

Start from the pattern that fails under load: do everything inside the request path.

import time
 
 
def fake_model(prompt: str) -> str:
    time.sleep(0.05)  # stand-in for network + generation
    return f"answer:{prompt[:20]}"
 
 
def run_agent_inline(user_goal: str) -> str:
    # In real agents this is many model/tool turns
    return fake_model(user_goal)
 
 
print(run_agent_inline("summarize ticket 42"))

Inline runs couple HTTP latency to every model and tool turn.
Under concurrent users, each request opens more provider load with no shared budget.
Scaling starts by decoupling intake from execution.

Related: What Actually Bottlenecks an Agent System at Scale

2. Put Agent Jobs on a Queue

Enqueue work, then process it separately.

from dataclasses import dataclass, field
from queue import Queue
from typing import Any
import uuid
 
 
@dataclass
class AgentJob:
    job_id: str
    goal: str
    meta: dict[str, Any] = field(default_factory=dict)
 
 
jobs: Queue[AgentJob | None] = Queue()
 
 
def enqueue(goal: str) -> str:
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    jobs.put(job)
    return job.job_id
 
 
job_id = enqueue("summarize ticket 42")
print("enqueued", job_id, "depth", jobs.qsize())

The API (or CLI) only accepts work and returns a job id.
A worker owns the slow loop.
Depth is your first capacity signal.

3. Process Jobs With One Sequential Worker

Drain the queue one job at a time.

def process_job(job: AgentJob) -> dict:
    result = fake_model(job.goal)
    return {"job_id": job.job_id, "result": result, "status": "done"}
 
 
def worker_loop(q: Queue[AgentJob | None]) -> None:
    while True:
        job = q.get()
        if job is None:
            q.task_done()
            break
        print(process_job(job))
        q.task_done()
 
 
enqueue("goal-a")
enqueue("goal-b")
jobs.put(None)  # poison pill to stop demo
worker_loop(jobs)

Sequential processing is a natural concurrency limit of 1.
Many production agent fleets start here and only raise concurrency when metrics allow.
Prefer durable brokers later; the shape stays the same.

4. Store Results Outside the Worker Memory

Clients need a place to poll status.

from threading import Lock
 
results: dict[str, dict] = {}
_lock = Lock()
 
 
def process_and_store(job: AgentJob) -> None:
    out = process_job(job)
    with _lock:
        results[job.job_id] = out
 
 
def get_status(job_id: str) -> dict:
    with _lock:
        return results.get(job_id, {"job_id": job_id, "status": "pending"})
 
 
j1 = enqueue("status demo")
# In a real system the worker runs in another process
process_and_store(jobs.get())
print(get_status(j1))

Never require the original HTTP connection to stay open for multi-minute agents.
Persist results (Redis, Postgres, object store) so restarts do not lose outcomes.
Return pending / running / done / failed with a stop reason when applicable.

5. Cap How Many Jobs Start at Once

Raise throughput carefully with a semaphore.

from concurrent.futures import ThreadPoolExecutor
from threading import Semaphore
 
MAX_IN_FLIGHT = 2
sem = Semaphore(MAX_IN_FLIGHT)
 
 
def run_with_cap(job: AgentJob) -> dict:
    with sem:
        return process_job(job)
 
 
def drain_with_pool(pending: list[AgentJob]) -> list[dict]:
    with ThreadPoolExecutor(max_workers=4) as pool:
        # max_workers can be higher than MAX_IN_FLIGHT; sem is the real gate
        return list(pool.map(run_with_cap, pending))
 
 
batch = [AgentJob(str(i), f"goal-{i}") for i in range(5)]
print(drain_with_pool(batch))

Pool size is not the same as allowed concurrent agent runs.
Cap against provider RPM/TPM and tool bulkheads, not only CPU.
Full recipe: Concurrency Limits and Backpressure for Agent Workers.

Intermediate Examples

6. Reject New Work When the Queue Is Too Deep

Backpressure starts at admission.

MAX_DEPTH = 3
 
 
def enqueue_with_limit(goal: str, q: Queue[AgentJob], max_depth: int = MAX_DEPTH) -> str:
    if q.qsize() >= max_depth:
        raise RuntimeError("queue_full: try later or shed load")
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal)
    q.put(job)
    return job.job_id
 
 
q2: Queue[AgentJob] = Queue()
for g in ["a", "b", "c", "d"]:
    try:
        print("ok", enqueue_with_limit(g, q2))
    except RuntimeError as e:
        print("rejected", e)

Unbounded queues only hide overload until latency explodes.
Map queue_full to HTTP 429/503 with Retry-After in real APIs.
Pair with cheaper-model degrade under pressure when the product allows it.

7. Prefer Fake Cache Hits Before Spending a Model Call

Caching is free capacity.

import hashlib
 
cache: dict[str, str] = {}
 
 
def cache_key(goal: str) -> str:
    return hashlib.sha256(goal.strip().lower().encode()).hexdigest()
 
 
def run_cached(goal: str) -> tuple[str, bool]:
    key = cache_key(goal)
    if key in cache:
        return cache[key], True
    value = fake_model(goal)
    cache[key] = value
    return value, False
 
 
print(run_cached("FAQ: reset password"))
print(run_cached("FAQ: reset password"))  # hit

Cache only deterministic, safe, non-personalized results unless you key by tenant and auth scope.
Tool-result caches often save more money than final-answer caches.
Expand in Caching Agent Responses and Tool Results to Cut Redundant Calls.

8. Sketch Cost Pressure Routing

When spend pressure is high, choose a cheaper model path.

from dataclasses import dataclass
 
 
@dataclass
class Budget:
    spent: float = 0.0
    soft_limit: float = 1.0
 
 
def pick_model(budget: Budget) -> str:
    if budget.spent >= budget.soft_limit * 0.8:
        return "cheap-small"
    return "premium-large"
 
 
def run_with_routing(goal: str, budget: Budget) -> dict:
    model = pick_model(budget)
    # pretend costs
    cost = 0.02 if model == "cheap-small" else 0.15
    budget.spent += cost
    return {"model": model, "result": fake_model(f"{model}:{goal}"), "spent": budget.spent}
 
 
b = Budget(spent=0.7, soft_limit=1.0)
print(run_with_routing("hard task", b))
print(run_with_routing("easy task", b))

Cost-aware scale is a scaling strategy, not only a billing feature.
Always eval quality when you demote models under load.
Deep dive: Cost-Aware Scaling: Routing Load to Cheaper Models Under Pressure.

What You Learned

Decouple intake from the agent loop with a queue and job ids.
Process with workers; store results for polling or push notify.
Cap in-flight runs separately from thread/process pool size.
Apply admission control when depth is high; cache and cheaper routes free capacity.

Next Steps

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.

Scaling Agent Systems Basics

Search Documentation

Scaling Agent Systems Basics