Background Agents

BackgroundAgentManager runs your agents away from any request: on demand, once at a set time, every N seconds, or on a cron schedule. Every run is recorded (its status, attempts, events and answer) in a task store, and with a durable store a run queued by one process is finished by another.

Every output on this page is what the code printed when it was run. The model’s wording, the ids and the times will differ on your run.

A first background run

import asyncio
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


async def main():
    agent = OmniCoreAgent(
        name="reporter",
        system_instruction="You write short status notes. Answer in plain text.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )
    manager = BackgroundAgentManager()
    await manager.register_agent("reporter", agent)
    await manager.register_task(
        task_id="daily_note",
        agent_id="reporter",
        query="In one sentence, remind the team to review open pull requests.",
        schedule={"type": "manual"},
        timeout_seconds=120,
    )

    run = await manager.run_now("daily_note", wait=True)
    print(run.status.value, "| attempt", run.attempt, "| answer:", run.result_preview)

    workspace = await manager.get_run_workspace(run.run_id)
    for item in workspace["files"]:
        print("file:", item["path"])

    for event in await manager.get_run_events(run.run_id):
        print(event["sequence"], event["event"])

    story = await agent.get_run_trajectory(run.run_id)
    for step in story["segments"][0]["trajectory"]["steps"]:
        for call in step["tool_calls"]:
            print("tool:", call["tool_name"], call["outcome"])
    await manager.shutdown()


asyncio.run(main())

completed | attempt 1 | answer: Please review the open pull requests.
file: background/reporter/daily_note/run_77ca668e7bb440e3a62909c70679d46f/output.md
file: background/reporter/daily_note/run_77ca668e7bb440e3a62909c70679d46f/run.json
file: background/reporter/daily_note/run_77ca668e7bb440e3a62909c70679d46f/events.jsonl
1 background_run_queued
2 background_run_claimed
3 background_run_started
4 background_run_heartbeat
5 background_run_completed
tool: write_file success

register_agent names an agent; register_task says which agent runs which query, and when.
run_now(..., wait=True) queued the run and, as no worker was started, ran it here and returned it finished.
Each run gets its own folder in the workspace: the query reaches the agent with the folder’s path and asks for the final result in output.md, notes in scratchpad/, logs in logs/. The manager writes run.json (the run) and events.jsonl (its events) beside it. The model chose to call write_file here; ask for “no files” in the instruction if you only want the answer.
result_preview is the agent’s answer, cut at 1,000 characters.
The agent’s run id is the background run id, so agent.get_run_trajectory(run_id) shows what the model did, step by step (Durable runs).

Schedules

import asyncio
from datetime import datetime, timezone
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


async def main():
    agent = OmniCoreAgent(
        name="ticker",
        system_instruction="Answer in plain text, in a few words. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )
    manager = BackgroundAgentManager()
    await manager.register_agent("ticker", agent)

    registered = datetime.now(timezone.utc)
    await manager.register_task(
        task_id="every_20s",
        agent_id="ticker",
        query="Say hello.",
        schedule={"type": "interval", "seconds": 20},
    )
    await manager.register_task(
        task_id="weekday_morning",
        agent_id="ticker",
        query="Write the morning summary.",
        schedule={"type": "cron", "expression": "0 9 * * 1-5", "timezone": "Africa/Lagos"},
    )
    for task_id in ("every_20s", "weekday_morning"):
        status = await manager.get_task_status(task_id)
        print(task_id, "next due:", status["schedule_state"]["next_due_at"])

    await manager.start()
    await asyncio.sleep(50)
    await manager.shutdown()

    for run in await manager.list_runs(task_id="every_20s"):
        due = (run.due_at - registered).total_seconds()
        print(run.trigger_type.value, run.status.value, f"due {due:.0f}s after registering", "|", run.result_preview)


asyncio.run(main())

every_20s next due: 2026-09-27T23:03:45.097687Z
weekday_morning next due: 2026-09-28T08:00:00Z
interval completed due 20s after registering | Hello!
interval completed due 40s after registering | Hello!

It was 23:03:25 UTC; 09:00 on Monday in Lagos is 08:00 UTC. Schedules run only while a manager is started: start() runs the worker loop until shutdown().

`schedule`	Fields	First run
`{"type": "manual"}`	none	only when you call `run_now`
`{"type": "once", "run_at": ...}`	`run_at` (a `datetime`; no time zone means UTC)	at `run_at`
`{"type": "interval", "seconds": 3600}`	`seconds` > 0	one interval after registering (or after `start_at`), not at once
`{"type": "cron", "expression": "0 9 * * 1-5"}`	five fields; `timezone` (default `UTC`)	the next matching minute

Every schedule but manual also takes start_at, end_at, jitter_seconds (a fixed offset, the same for every worker) and misfire_policy: what happens to times missed while no manager was running. run_once (the default) queues one run for the missed times, then continues from now; skip_missed runs nothing for a time missed by more than a minute and waits for the next; queue_all queues a run for each one missed. Want an interval task to run at once too? Call run_now after registering. pause_task stops a schedule (runs already queued go on) and resume_task starts it again; run_now still works on a paused task.

Overlap: a task that is due while it is running

import asyncio
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


async def main():
    agent = OmniCoreAgent(
        name="ticker",
        system_instruction="Answer in plain text, in a few words. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )
    manager = BackgroundAgentManager()
    await manager.register_agent("ticker", agent)
    await manager.register_task(
        task_id="hello",
        agent_id="ticker",
        query="Say hello.",
        schedule={"type": "manual"},
        overlap_policy="skip_if_running",  # the default
    )
    await manager.start()
    first = await manager.run_now("hello")
    second = await manager.run_now("hello")
    print("first:", first.status.value, "| second:", second.status.value)
    first = await manager.wait_for_run(first.run_id, timeout_seconds=60)
    print("first:", first.status.value, "|", first.result_preview)
    await manager.shutdown()


asyncio.run(main())

first: queued | second: skipped
first: completed | Hello!

`overlap_policy`	The new run, while one is active
`skip_if_running` (default)	is recorded as `skipped` and never runs
`queue_next`	waits in the queue until the earlier run ends
`cancel_previous`	runs; the earlier one is cancelled: a queued run never starts, a running one is interrupted within a heartbeat (a quarter of the lease) and recorded `cancelled`
`allow_parallel`	runs as soon as a worker is free

A run waiting for a person (below) still holds its task: under skip_if_running, new runs are skipped until it is decided.

One manager runs one run at a time. allow_parallel lets a task’s runs go side by side only when several workers share the store (below).

Retries and timeouts

import asyncio
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


async def main():
    agent = OmniCoreAgent(
        name="writer",
        system_instruction="Answer in plain text. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )
    manager = BackgroundAgentManager()
    await manager.register_agent("writer", agent)
    await manager.register_task(
        task_id="long_essay",
        agent_id="writer",
        query="Write a 1500-word essay on the history of the printing press.",
        schedule={"type": "manual"},
        timeout_seconds=3,
        retry_policy={"max_retries": 1, "initial_delay_seconds": 2},
    )
    run = await manager.run_now("long_essay", wait=True, timeout_seconds=60)
    print(run.status.value, "| attempts:", run.attempt, "| error:", run.error)
    for attempt in await manager.list_attempts(run.run_id):
        print(" ", attempt.attempt_number, attempt.reason.value, attempt.status.value, attempt.retry_delay_seconds)
    await manager.shutdown()


asyncio.run(main())

timeout | attempts: 2 | error: Run exceeded its 3 second deadline
  1 initial timeout 2
  2 retry timeout None

timeout_seconds on the task is each attempt’s deadline (none by default). An attempt past it ends timeout.
timeout_seconds on run_now (and wait_for_run, run_until_terminal) is only how long you wait; the run is returned as it is then.
An attempt that raises, or whose agent run ends with an error, or that times out, is retried while retries are left: the run goes retrying, then queued again after the delay. When none are left it ends failed or timeout, with the error. A retry is a new attempt from the first step, not a resume: the earlier attempt’s tool calls can run again, so retry only tasks whose tools are safe to repeat (see what is guaranteed).

`retry_policy` key	Default
`max_retries`	`0`	retries after the first attempt
`initial_delay_seconds`	`30`	wait before a retry
`backoff`	`"fixed"`	`"exponential"` doubles the delay each retry
`max_delay_seconds`	`300`	the longest delay
`retry_on`	`["exception", "timeout"]`	which failures are retried

run_now(wait=True) returns when the run ends, at timeout_seconds, or when an attempt failed and its retry is scheduled for later: the run comes back queued, with its queued_at, rather than holding you through the whole backoff. A started manager runs the retry when it is due; on a manager that is not started, nothing does, so start() it when a task retries with a delay.

A run that waits for a person

A background agent under governance can pause for an approval or a budget. The run then waits, holding no worker, until someone decides on the agent and resume_run queues it again. Its next attempt continues the same run; it does not start over.

import asyncio
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent, ToolRegistry
from omnicoreagent.governance import PolicyEffect, PolicyRule, build_default_policy

tools = ToolRegistry()


@tools.register_tool("issue_refund")
def issue_refund(order_id: str, amount: float) -> dict:
    """Refund an order."""
    return {"order_id": order_id, "refunded": amount}


policy = build_default_policy("interactive-dev")
policy.rules.ask.append(
    PolicyRule(rule_id="ask_before_refunds", effect=PolicyEffect.ASK,
               capability="tool.local.call", target={"tool_name": "issue_refund"})
)


async def main():
    agent = OmniCoreAgent(
        name="support",
        system_instruction="You help with orders. Refund when asked. Answer in plain text, in one sentence. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
        local_tools=tools,
        agent_config={"governance_config": {"enabled": True, "policy": policy}},
    )
    manager = BackgroundAgentManager()
    await manager.register_agent("support", agent)
    await manager.register_task(
        task_id="refund_1042",
        agent_id="support",
        query="Please refund order 1042 in full: 42 dollars.",
        schedule={"type": "manual"},
    )

    run = await manager.run_now("refund_1042", wait=True, timeout_seconds=120)
    print(run.status.value, "|", run.result_preview)

    # A person decides, from the agent's run record.
    for approval in (await agent.get_run(run.run_id))["approvals"]:
        print("asks:", approval["tool_name"], approval["arguments"])
        await agent.resolve_approval(run.run_id, approval["approval_id"],
                                     decision="approve", approver="alice")

    await manager.resume_run(run.run_id)  # queued again, same run_id
    run = await manager.run_until_terminal(run.run_id, timeout_seconds=120)
    print(run.status.value, "| attempt", run.attempt, "|", run.result_preview)

    story = await agent.get_run_trajectory(run.run_id)
    print("segments:", [segment["status"] for segment in story["segments"]])
    print("events:", [e["event"] for e in await manager.get_run_events(run.run_id)])
    await manager.shutdown()


asyncio.run(main())

awaiting_approval | Waiting for approval: issue_refund
asks: issue_refund {'order_id': '1042', 'amount': 42}
completed | attempt 2 | Refunded order 1042 in full for $42.
segments: ['suspended', 'completed']
events: ['background_run_queued', 'background_run_claimed', 'background_run_started', 'background_run_heartbeat', 'background_run_awaiting_approval', 'background_run_queued', 'background_run_claimed', 'background_run_started', 'background_run_completed']

A budget works the same way. With "budgets": {"request": [{"meter": "model_cost_usd", "limit": 0.0005}]} and max_tokens 300, a one-line tagline task printed:

run = await manager.run_now("tagline", wait=True, timeout_seconds=120)
print(run.status.value, "|", run.result_preview)

await agent.grant_budget(run.run_id, approver="alice", amount=0.01, note="one tagline")
await manager.resume_run(run.run_id)
run = await manager.run_until_terminal(run.run_id, timeout_seconds=120)
print(run.status.value, "|", run.result_preview)

awaiting_budget | Waiting for budget: request model_cost_usd (needs 0.001336 more)
completed | Freshly baked warmth in every bite.

Decide with the agent: resolve_approval, grant_budget or deny_budget. After a denial, resume_run lets the run end cleanly.
cancel_run(run_id) ends a waiting run instead: it is cancelled, and so is the agent’s own record of it.
With the manager started, resume_run is enough: a worker picks the run up. run_until_terminal runs it here, for a manager that is not started.
A scheduled task’s runs share one session by default (session_policy {"mode": "task"}), so a session budget with the default total window adds up across runs and, once spent, pauses (by default) every run after it. For a limit per run, use a request budget, or give each run its own session with session_policy={"mode": "run"}.
In another process, build the agent with the same memory store to decide (Approvals: from another process).

Reading a run

You want	Call
status, attempt count, times, answer, error	`get_run(run_id)`
every attempt: worker, reason (`initial`, `retry`, `recovery`, or `lease_expired` for an attempt whose worker’s lease ran out), how it ended	`list_attempts(run_id)`
lifecycle events, numbered from 1	`get_run_events(run_id)`
the run’s folder and files	`get_run_workspace(run_id)`
what the model did: steps, calls, tokens, cost, across pauses and attempts	`agent.get_run_trajectory(run_id)`
one task: schedule state, next due time, run counts by status	`get_task_status(task_id)`
the whole manager	`get_manager_status()`, `list_runs(task_id=..., status=...)`

Events come from the manager process that emitted them, or from the run’s events.jsonl in the workspace, so a finished run’s events can still be read after a restart. A run’s statuses are queued, claimed, running, retrying; it ends completed, failed, cancelled, timeout or skipped, or waits in awaiting_approval or awaiting_budget. The events page lists every event name.

Task stores

The task store keeps agents’ specs, tasks, schedule state, runs, attempts, leases and cancel flags. It is not the agent’s memory: conversations and run records stay in the agent’s memory store.

`task_store=`	Keeps state	Install
none (`"in_memory"`)	the in-memory store: while the process lives	core
`"sql"`	SQLite at `.omnicoreagent/background.db`, or `DATABASE_URL` when set	`omnicoreagent[postgres]` (it ships SQLAlchemy, which SQLite needs too)
`{"backend": "sql", "url": "postgresql://user:pass@host/db"}`	PostgreSQL (or `mysql+pymysql://...`)	`omnicoreagent[postgres]`
`"redis"` or `{"backend": "redis", "url": "redis://localhost:6379/0"}`	Redis (`"redis"` reads `REDIS_URL`)	`omnicoreagent[redis]`
`"mongodb"` or `{"backend": "mongodb", "uri": "mongodb://localhost:27017", "database": "omnicoreagent"}`	MongoDB (`"mongodb"` reads `MONGODB_URI`)	`omnicoreagent[mongodb]`

Use SQL for local durability with SQLite, or PostgreSQL or MySQL when several processes share one store. Use Redis when your deployment already operates Redis, with persistence on and no eviction of task-store keys. Use MongoDB when MongoDB is your durable operational store. A dict also takes prefix (tables or keys, so deployments can share one database) and connect_timeout. Background run history is not pruned.

A queued run survives the process

With a durable store, a queued run left by one process is finished by another over the same task store. Two scripts share one definition:

# jobs.py
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


def build_agent() -> OmniCoreAgent:
    return OmniCoreAgent(
        name="reporter",
        system_instruction="You write short status notes. Answer in plain text. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )


def build_manager() -> BackgroundAgentManager:
    return BackgroundAgentManager(
        task_store={"backend": "sql", "url": "sqlite:///./background.db"}
    )

Process one queues a run and exits

# enqueue.py
import asyncio
from jobs import build_agent, build_manager


async def main():
    manager = build_manager()
    await manager.register_agent("reporter", build_agent())
    await manager.register_task(
        task_id="release_note", agent_id="reporter",
        query="In one sentence, announce that version 2 is out.",
        schedule={"type": "manual"},
    )
    run = await manager.run_now("release_note")  # queued; no worker here
    print(run.run_id, run.status.value)
    await manager.shutdown()


asyncio.run(main())

run_abfd26183727401e919e69c6cdf9d983 queued

Process two finds it and runs it

# worker.py
import asyncio
from jobs import build_agent, build_manager


async def main():
    manager = build_manager()
    await manager.register_agent("reporter", build_agent(), replace=True)
    (queued,) = await manager.list_runs(status="queued")
    print("found:", queued.run_id, queued.status.value)

    await manager.start()
    run = await manager.wait_for_run(queued.run_id, timeout_seconds=120)
    print(run.status.value, "|", run.result_preview)
    await manager.shutdown()


asyncio.run(main())

found: run_abfd26183727401e919e69c6cdf9d983 queued
completed | Version 2 is out.

The store keeps the agent’s spec (model, instruction, settings, MCP servers), so replace=True is needed to register it again; a worker that never registers it rebuilds the agent from that spec, without local tools. Register in every process an agent that has local tools.

Several workers, one store

Each started manager is a worker. Workers on one store claim runs with a lease (30 seconds, renewed as the run goes), so no run is taken twice. Here four runs of an allow_parallel task, queued into Redis, then two worker processes started side by side:

# jobs.py
from omnicoreagent import BackgroundAgentManager, OmniCoreAgent


def build_agent() -> OmniCoreAgent:
    return OmniCoreAgent(
        name="translator",
        system_instruction="Translate into French. Answer in plain text only. Do not write files.",
        model_config={"provider": "openai", "model": "gpt-5.4-mini"},
    )


def build_manager(worker_id: str | None = None) -> BackgroundAgentManager:
    return BackgroundAgentManager(
        task_store={"backend": "redis", "url": "redis://localhost:6379/0"},
        worker_id=worker_id,  # a name for this worker; a random one otherwise
    )

# enqueue.py: register, then queue four runs, each with its own query
import asyncio
from jobs import build_agent, build_manager


async def main():
    manager = build_manager()
    # replace=True: the store keeps both from an earlier run of this script.
    await manager.register_agent("translator", build_agent(), replace=True)
    await manager.register_task(
        task_id="translate",
        agent_id="translator",
        query="Good morning.",
        schedule={"type": "manual"},
        overlap_policy="allow_parallel",  # the default skips a run while one is active
        replace=True,
    )
    for text in ["Good morning.", "Thank you.", "See you tomorrow.", "Where is the station?"]:
        await manager.run_now("translate", query=text)
    await manager.shutdown()


asyncio.run(main())

# worker.py: run as `python worker.py worker-a` and `python worker.py worker-b`
import asyncio, sys
from jobs import build_agent, build_manager


async def main(worker_id: str):
    manager = build_manager(worker_id)
    await manager.register_agent("translator", build_agent(), replace=True)
    await manager.start()
    await asyncio.sleep(40)
    await manager.shutdown()


asyncio.run(main(sys.argv[1]))

# report.py
import asyncio
from jobs import build_manager


async def main():
    m = build_manager()
    for run in await m.list_runs(task_id="translate"):
        at = await m.list_attempts(run.run_id)
        print([(a.worker_id, a.status.value, a.started_at.strftime('%H:%M:%S'),
                a.finished_at and a.finished_at.strftime('%H:%M:%S')) for a in at],
              run.status.value, run.result_preview, run.error)
    await m.shutdown()


asyncio.run(main())

[('worker-a', 'completed', '23:10:52', '23:11:11')] completed Bonjour. None
[('worker-b', 'completed', '23:10:53', '23:11:11')] completed Veuillez fournir le texte à traduire en français. None
[('worker-a', 'completed', '23:11:11', '23:11:13')] completed À demain. None
[('worker-b', 'completed', '23:11:12', '23:11:13')] completed Où se trouve la station ? None

Each worker took two runs, side by side. One run asked for the text instead of translating it: runs of one task share a memory session by default (session_policy {"mode": "task"}), so parallel runs read each other’s turns. For runs that must not see each other, give the task session_policy={"mode": "run"}. If a worker dies, its lease runs out and another worker takes the run back, continuing from the agent’s checkpoint; list_attempts shows the lost attempt with the reason lease_expired and the attempt that took over with recovery. More in Stores and scale.

Starting and shutting down

start() prepares the store, loads each registered agent’s model client (so the first run does not pay for that import), and starts the worker loop. Each turn it takes back runs whose lease ran out, queues due schedules, and runs one queued run.
Without start(), nothing is scheduled; run_now(wait=True) and run_until_terminal run a queued run in your process.
shutdown() stops the loop, stops the runs it is executing, and closes the store. Each stopped attempt is recorded as failed with the error worker shutdown. A task with retries left puts the run back in the queue (queued, for the next worker); otherwise the run ends failed with that error. It is not cancelled: only cancel_run cancels. Call shutdown() before the process exits.

With a governance_engine on the manager, a task is bound to the policy it was created under. When the policy changes, the task can still be paused or deleted but not run: at its next due time its schedule is paused, and get_task_status shows why in schedule_state.paused_reason. Any change to the policy counts, a budget number included. Register the task again with replace=True (over HTTP, POST /background/tasks with "replace": true): the new policy decides whether it may (it needs background.task.update), binds it to that policy, and its schedule runs again. A task a person paused stays paused.

Over HTTP

OmniServe registers the served agent with a background manager and exposes it under /background/tasks/:

curl -X POST http://localhost:8000/background/tasks \
  -H "Content-Type: application/json" \
  -d '{"task_id": "health_report", "query": "Check system health and write the report.",
       "schedule": {"type": "manual"}, "timeout_seconds": 60}'
curl -X POST http://localhost:8000/background/tasks/health_report/run \
  -H "Content-Type: application/json" -d '{"wait": false}'

curl http://localhost:8000/background/status
curl http://localhost:8000/background/runs/$RUN_ID
curl http://localhost:8000/background/runs/$RUN_ID/events
curl http://localhost:8000/background/runs/$RUN_ID/workspace

Every route, and the task-store environment variables, are on the OmniServe page.

Options

register_task takes:

Option	Default	What it does
`task_id`, `agent_id`	required	letters, numbers, `_`, `-`, `.`
`query`	required	what the agent is asked; `run_now(query=...)` overrides it once
`schedule`	required	manual, once, interval or cron (above)
`timeout_seconds`	none	each attempt’s deadline
`retry_policy`	no retries	above
`overlap_policy`	`"skip_if_running"`	above
`session_policy`	`{"mode": "task"}`	the memory session: `task` (one per task, runs remember each other), `run` (a new one per run), or `fixed` with a `session_id`
`workspace_policy`	`background/{agent_id}/{task_id}/{run_id}`	the run folder’s `namespace_template`; `write_run_json`, `write_events_jsonl`
`enabled`	`True`	`False` stops the schedule and `run_now`
`replace`	`False`	overwrite a task with the same id

BackgroundAgentManager(...) takes task_store, memory_router, workspace, telemetry_store, governance_engine, worker_id and lease_seconds (30). Change a task later with update_task(task_id, patch). Every method is in the BackgroundAgentManager reference.

When things go wrong

AgentNotFoundError: Agent not found

The task names an agent not registered in this store:

AgentNotFoundError: Agent not found: nobody

Call register_agent first, with the same id.

AgentAlreadyRegisteredError / TaskAlreadyRegisteredError

The id is already in the store; with a durable store, from an earlier process:

AgentAlreadyRegisteredError: Agent already registered: a
TaskAlreadyRegisteredError: Task already registered: t

Pass replace=True to overwrite it.

TaskNotFoundError: Task not found

run_now on a task that does not exist, or has enabled=False:

TaskNotFoundError: Task not found: nope

ValidationError: a schedule that cannot run

Each schedule type checks its fields:

Value error, interval schedule requires seconds > 0
Value error, cron expression must have five fields
Value error, once schedule requires run_at
Value error, unknown schedule timezone: Lagos

Time zones are IANA names: Africa/Lagos, not Lagos. Ids are checked too: task_id must contain only letters, numbers, '_', '-', and '.'.

ValueError: only a run in awaiting_approval or awaiting_budget can be resumed

resume_run is for a waiting run:

ValueError: Run run_dc1ac333fd374fbb80b46ef6292ac5d3 is completed; only a run in awaiting_approval or awaiting_budget can be resumed

To run a finished task again, call run_now.

InvalidTaskStoreError

An unknown backend, or one missing its address:

InvalidTaskStoreError: task_store must be 'in_memory', 'sql', 'redis', or 'mongodb'
Value error, redis task store requires url
Value error, mongodb task store requires uri and database

PostgreSQL is "sql" with a postgresql:// URL.

My second run came back skipped

The default overlap_policy is skip_if_running, and a run of the task was still active, or waiting for a person. Use queue_next, or decide the waiting run.

An interval task has not run yet

Its first run is one interval after it was registered, and only while a manager is started. get_task_status(task_id)["schedule_state"]["next_due_at"] says when.

BackgroundAgentManager reference

Every method, run status and overlap policy.

Durable runs

Run records, trajectories, resuming after a crash.

Stores and scale

Choosing stores and running several workers.

OmniServe

Background tasks over HTTP.

Agent in a container

Package the agent and its worker.

Approvals

Deciding what a paused run asked for.

Get Started

Make It Safe

Run It

See and Improve

Build

How It Works

Reference

HTTP API (OmniServe)

Releases

Background Agents

Background Agents

A first background run

Schedules

Overlap: a task that is due while it is running

Retries and timeouts

A run that waits for a person

Reading a run

Task stores

A queued run survives the process

Several workers, one store

Starting and shutting down

Over HTTP

Options

When things go wrong

Next

BackgroundAgentManager reference

Durable runs

Stores and scale

OmniServe

Agent in a container

Approvals

​Background Agents

​A first background run

​Schedules

​Overlap: a task that is due while it is running

​Retries and timeouts

​A run that waits for a person

​Reading a run

​Task stores

​A queued run survives the process

​Several workers, one store

​Starting and shutting down

​Over HTTP

​Options

​When things go wrong

​Next

BackgroundAgentManager reference

Durable runs

Stores and scale

OmniServe

Agent in a container

Approvals

Background Agents

A first background run

Schedules

Overlap: a task that is due while it is running

Retries and timeouts

A run that waits for a person

Reading a run

Task stores

A queued run survives the process

Several workers, one store

Starting and shutting down

Over HTTP

Options

When things go wrong

Next