Batch and durability

Bound concurrent batches, choose in-process or durable background runs, and schedule AgentOS endpoints.

A single agent.run(files=[File(...)]) handles one document. Folders, queues, and nightly drops also need concurrency limits, retry policy, and idempotent writes. The patterns below cover bounded async batches, in-process background runs, durable AgentOS queues, and scheduled endpoints.

Concurrent batch over a list

The simplest batch is a folder of files. agent.arun is async, so a semaphore plus asyncio.gather is enough.

import asyncio
from pathlib import Path

from agno.agent import Agent
from agno.media import File
from agno.models.openai import OpenAIResponses

from your_schemas import Invoice  # define your output schema once


agent = Agent(
    model=OpenAIResponses(id="gpt-5.5"),
    instructions="Extract invoice fields and line items. Null for missing.",
    output_schema=Invoice,
)


async def extract_one(path: Path, sem: asyncio.Semaphore) -> Invoice:
    async with sem:
        run = await agent.arun(
            "Extract this invoice.",
            files=[File(filepath=str(path))],
        )
        return run.content


async def extract_folder(folder: Path, concurrency: int = 8) -> list[Invoice]:
    sem = asyncio.Semaphore(concurrency)
    paths = sorted(folder.glob("*.pdf"))
    return await asyncio.gather(*(extract_one(p, sem) for p in paths))


invoices = asyncio.run(extract_folder(Path("./incoming-invoices")))
# [Invoice(invoice_number='1042', ...), Invoice(invoice_number='1043', ...), ...]

A semaphore bounds in-flight calls and memory use. It does not enforce requests-per-minute or token quotas. Add a rate limiter and retry policy for provider limits, and choose concurrency based on the model quota and downstream write capacity.

Background runs for long jobs

background=True returns a pending run while an asyncio task continues in the current process. Poll the persisted run state when the caller should not wait for the model response.

import asyncio

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.media import File
from agno.models.openai import OpenAIResponses
from agno.run.base import RunStatus

from your_schemas import Invoice

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")
agent = Agent(model=OpenAIResponses(id="gpt-5.5"), db=db, output_schema=Invoice)


async def extract_long(file_url: str) -> Invoice:
    started = await agent.arun(
        "Extract this invoice.",
        files=[File(url=file_url)],
        background=True,
    )
    # started.status is RunStatus.pending; the work continues in the background.

    deadline = asyncio.get_running_loop().time() + 600
    while asyncio.get_running_loop().time() < deadline:
        await asyncio.sleep(2)
        run = await agent.aget_run_output(
            run_id=started.run_id,
            session_id=started.session_id,
        )
        if run is None:
            continue
        if run.status == RunStatus.completed:
            return Invoice.model_validate(run.content)
        if run.status == RunStatus.error:
            raise RuntimeError(f"Run {started.run_id} failed")
        if run.status in (RunStatus.cancelled, RunStatus.paused):
            raise RuntimeError(f"Run {started.run_id} stopped with {run.status}")

    raise TimeoutError(f"Run {started.run_id} did not finish within 10 minutes")

Content loaded from the database comes back as a plain dict, so validate it back into your schema. The database persists pending, running, and terminal state for polling. The work itself is an in-process asyncio task. If the process exits, that direct agent.arun() task does not resume from the run record. Submit through an AgentOS durable queue when execution must survive process failure.

Durable background runs

durable_intake.py

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS, QueueConfig

from your_schemas import Invoice

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

extractor = Agent(
    id="invoice-extractor",
    model=OpenAIResponses(id="gpt-5.5"),
    db=db,
    output_schema=Invoice,
)

agent_os = AgentOS(
    agents=[extractor],
    db=db,
    queue=QueueConfig(
        durable=True,
        max_concurrency=8,
        max_queue_depth=1000,
        max_attempts=1,
    ),
)
app = agent_os.get_app()

if __name__ == "__main__":
    agent_os.serve(app="durable_intake:app", port=7777)

In v3.0.4, the durable queue accepts runs that do not include uploaded files or media. Persist the document first, then send its URL or object key to an agent, team, or workflow whose code resolves that reference. Multipart file and media submissions use the in-process path. Use an external durable worker when the original upload must be the queued payload.

For an extractor configured to resolve a stored URL, submit a non-streaming background run:

curl -X POST http://localhost:7777/agents/invoice-extractor/runs \
  -F 'message=Process the invoice at s3://invoices/1042.pdf' \
  -F 'background=true' \
  -F 'stream=false'

AgentOS commits the accepted job to its database before returning 202, and a live replica can claim queued work after a restart. The default max_attempts=1 reports an interrupted claimed job as failed without silently repeating possible side effects. Set max_attempts=2 or higher for automatic crash retries after making the target idempotent.

Poll /agents/invoice-extractor/runs/{run_id}?session_id={session_id} for the result. See Durable Queue for submission, polling, retry, and dead-letter operations.

Scheduled batch with retries

For nightly intake (an SFTP drop, a Drive folder, a queue), put an AgentOS in front of your agent and let the scheduler fire the run on cron.

scheduled_intake.py

from agno.agent import Agent
from agno.db.postgres import PostgresDb
from agno.models.openai import OpenAIResponses
from agno.os import AgentOS

from your_schemas import Invoice

db = PostgresDb(db_url="postgresql+psycopg://ai:ai@localhost:5532/ai")

extractor = Agent(
    id="invoice-extractor",
    model=OpenAIResponses(id="gpt-5.5"),
    db=db,
    output_schema=Invoice,
)

agent_os = AgentOS(
    agents=[extractor],
    db=db,
    scheduler=True,
    scheduler_poll_interval=15,    # check for due jobs every N seconds
)
app = agent_os.get_app()

In the same scheduled_intake.py module, create the schedule before starting the server. ScheduleManager writes to the same db the AgentOS polls.

from agno.scheduler import ScheduleManager

mgr = ScheduleManager(db)

schedule = mgr.create(
    name="nightly-invoice-intake",
    cron="0 2 * * *",                 # 2am every day
    endpoint="/agents/invoice-extractor/runs",
    payload={"message": "Process the overnight invoice drop."},
    timezone="America/New_York",
    max_retries=2,
    retry_delay_seconds=300,
    if_exists="update",
)

if __name__ == "__main__":
    agent_os.serve(app="scheduled_intake:app", port=7777)

if_exists="update" updates a schedule found by name instead of creating another schedule. Agno v3.0.4 enforces schedule-name uniqueness per owner. The lookup and write are separate operations, so simultaneous first-time creation can still race and one insert can fail with a uniqueness error. Run schedule bootstrap once or retry that conflict when multiple application workers can start concurrently. The option also does not make the scheduled endpoint idempotent. The executor retries failed HTTP or run attempts with the configured delay, and each attempt writes a row to agno_schedule_runs.

Make the endpoint idempotent before enabling retries. A response or polling failure can cause another attempt after the target already performed a side effect.

In Agno v3.0.4, a scheduler claim becomes stale after 300 seconds and the executor does not refresh the lock. A schedule that runs longer than five minutes can be claimed again. Keep the target idempotent and use an external scheduler or queue for long-running jobs.

Pattern comparison

Pattern	When to reach for it	Process lifetime
asyncio.gather over agent.arun	One-time backfill, a fixed list of files	One process, end-to-end
agent.arun(background=True) + poll	Single long document without blocking the caller	Task in the current process; status in db
AgentOS(queue=QueueConfig(durable=True))	Background runs that must survive restarts or move across replicas	Job and run state in db; worker in any live replica
AgentOS(scheduler=True) + ScheduleManager	Recurring intake (nightly, hourly)	Poller and execution in AgentOS; schedule and attempts in db
Workflow with Loop / Parallel steps	Multi-step pipelines per document	Either ad-hoc or scheduled

The scheduler fires endpoints. Endpoints are agents, teams, or workflows. So a nightly job that ingests a folder, extracts each file, and writes to your warehouse is a workflow exposed at /workflows/<id>/runs, scheduled with the same ScheduleManager.create call. See Workflows.

Observability

Every execution attempt creates a row in agno_schedule_runs with status and timing. run_id and session_id are present when the target run starts successfully. Inspect recent activity with:

runs = mgr.get_runs(schedule.id, limit=100)
for r in runs:
    print(r.triggered_at, r.status, r.attempt, r.error or "")

Failed attempts keep their error text. Retries are separate rows with the same schedule_id and an incrementing attempt. Monitor these rows and route exhausted failures to your operational queue.

Production checklist

Concern	What to add
Idempotency per document	Put a stable document ID under a unique constraint in the destination system. A repeated session_id scopes history; it does not deduplicate runs or writes.
Dead-letter queue	After retries are exhausted, query the final failed agno_schedule_runs row and enqueue it yourself.
Per-provider rate limiting	Combine the semaphore with a requests or token rate limiter and retry backoff.
Storage of inputs	File(url=...) keeps the URL but not the bytes. If retention matters, store the source PDF before extraction.
Cost reporting	RunMetrics.cost is populated when the provider returns it. Use provider billing data for reconciliation.

Next steps

Task	Guide
Pause on low-confidence fields	Human routing and eval
Compose multiple agents into a pipeline	Workflows
See the workflow + scheduler integration	Scheduling

Developer Resources

• Scheduler cookbook
• Background execution cookbook
• Workflows cookbook