> ## Documentation Index
> Fetch the complete documentation index at: https://agno-v2-codex-docs-audit-20260719-0149.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

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

```python theme={null}
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.

```python theme={null}
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.4 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.

```python scheduled_intake.py theme={null}
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.

```python theme={null}
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.4 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`.

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

<Warning>
  In Agno v2.7.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.
</Warning>

## 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](/workflows/overview).

## 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:

```python theme={null}
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](/use-cases/document-processing/human-routing-and-eval) |
| Compose multiple agents into a pipeline  | [Workflows](/workflows/overview)                                                |
| See the workflow + scheduler integration | [Scheduling](/features/scheduling)                                              |

## Developer Resources

* [Scheduler cookbook](https://github.com/agno-agi/agno/tree/v2.7.4/cookbook/05_agent_os/scheduler)
* [Background execution cookbook](https://github.com/agno-agi/agno/blob/v2.7.4/cookbook/02_agents/14_advanced/background_execution.py)
* [Workflows cookbook](https://github.com/agno-agi/agno/tree/v2.7.4/cookbook/04_workflows)