Batch and durability

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

Save the Invoice and LineItem definitions from Invoices and receipts in your_schemas.py beside the scripts on this page, and put input PDFs in incoming-invoices/. The simplest batch is a folder of files. agent.arun is async, so a semaphore plus asyncio.gather is enough.

Before running these examples, create and activate a Python environment, then install the provider and set its key:

uv venv .venv --python 3.12
source .venv/bin/activate
uv pip install "agno[openai]"
export OPENAI_API_KEY="..."

Replace every https://example.com/... PDF URL with a real PDF URL accessible to the model provider, or use File(filepath="...") for a local PDF. Create the referenced local files before running the example. Extraction quality depends on the document and model; the output comments are illustrative.

import asyncio
from pathlib import Path

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

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))],
        )
        if run.status != RunStatus.completed or not isinstance(run.content, Invoice):
            raise RuntimeError(f"No validated Invoice for {path}; route this file to review")
        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 malformed or failed extraction raises before it can enter the returned list[Invoice]; the batch then raises instead of returning a complete list. 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

For the PostgreSQL examples, install uv pip install "agno[os,postgres]" and start a writable PostgreSQL service. Replace the example database URL with your service URL.

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

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)

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.

The extractor above has no resolver: it only processes files supplied through files. Before using the submission below, add application code that authorizes the stored object reference, fetches the PDF, and attaches it as a File. A bare S3 URL in message does not do that automatically.

For that configured extractor, 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.

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()

This registration is a composition skeleton. Add a workflow or application resolver that enumerates the overnight folder, supplies each PDF through files, checks each extraction, and writes results idempotently. The message in the schedule below does not implement folder ingestion.

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 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.

Pattern comparison

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

Next steps

Developer Resources

• Scheduler cookbook
• Background execution cookbook
• Workflows cookbook