Batch and durability

Bound concurrent batches, poll in-process 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, and scheduled AgentOS 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, Agno v2.7.2 does not resume that task from the database. Use a durable external queue or worker system when execution must survive process failure.

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. The lookup and write are not atomic, and v2.7.2 does not place a unique constraint on schedule names. Run schedule bootstrap once 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 v2.7.2, 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(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