BackgroundAgentManager

from omnicoreagent import BackgroundAgentManager

manager = BackgroundAgentManager()                    # tasks kept in memory
manager = BackgroundAgentManager(task_store="sql")   # SQLite, or DATABASE_URL; needs omnicoreagent[postgres]

The guide is Background agents.

Run statuses

A run is queued, then claimed by a worker, running, and retrying between attempts. It ends in a terminal status, or pauses in a waiting one until resume_run; run_now(wait=True) and wait_for_run return at either.

Status
`queued`	active
`claimed`	active
`running`	active
`retrying`	active
`completed`	ends
`failed`	ends
`cancelled`	ends
`timeout`	ends
`skipped`	ends
`awaiting_approval`	waits
`awaiting_budget`	waits

Overlap policies

What a task does when it is due while one of its runs is still active (overlap_policy; the default is skip_if_running):

Policy
`skip_if_running`
`queue_next`
`cancel_previous`
`allow_parallel`

Methods

`cancel_run`

async def cancel_run(run_id: str) -> None: ...

Cancel a queued, running or waiting run.

`delete_task`

async def delete_task(task_id: str, delete_runs: bool = False) -> None: ...

Delete a task; delete_runs deletes its runs too.

`get_agent`

async def get_agent(agent_id: str) -> BackgroundAgentSpec | None: ...

The registered agent’s spec, or None.

`get_manager_status`

async def get_manager_status() -> dict[str, Any]: ...

Agents, tasks, active runs and run counts by status, for this manager’s store.

`get_run`

async def get_run(run_id: str) -> BackgroundRun | None: ...

The run: its status, attempts count, timings, result preview and error.

`get_run_events`

async def get_run_events(run_id: str) -> list[dict[str, Any]]: ...

The run’s lifecycle events (queued, claimed, started, completed, …), in order.

`get_run_workspace`

async def get_run_workspace(run_id: str) -> dict[str, Any]: ...

Return the durable workspace location and visible files for a run.

`get_task`

async def get_task(task_id: str) -> BackgroundTaskSpec | None: ...

The task, or None.

`get_task_status`

async def get_task_status(task_id: str) -> dict[str, Any]: ...

A task with its schedule state, active runs and a count of its runs by status.

`initialize`

async def initialize() -> None: ...

Prepare the task store. start and the first use call it for you.

`list_agents`

async def list_agents() -> list[BackgroundAgentSpec]: ...

Every registered agent’s spec.

`list_attempts`

async def list_attempts(run_id: str) -> list[BackgroundAttempt]: ...

Every attempt of a run, with how each ended.

`list_runs`

async def list_runs(task_id: str | None = None, status: str | RunStatus | None = None) -> list[BackgroundRun]: ...

Runs, optionally one task’s, optionally in one status.

`list_tasks`

async def list_tasks(agent_id: str | None = None) -> list[BackgroundTaskSpec]: ...

Every task, or one agent’s.

`pause_task`

async def pause_task(task_id: str) -> None: ...

Stop scheduling a task; runs already queued or running continue.

`recover_expired_runs`

async def recover_expired_runs() -> None: ...

Take back runs whose worker stopped renewing its lease (a process that died), so they retry or fail.

`register_agent`

async def register_agent(agent_id: str, agent: Any, replace: bool = False) -> BackgroundAgentSpec: ...

`register_agent_spec`

async def register_agent_spec(spec: BackgroundAgentSpec | dict[str, Any], replace: bool = False) -> BackgroundAgentSpec: ...

`register_task`

async def register_task(spec: BackgroundTaskSpec | dict[str, Any] | None = None, *, task_id: str | None = None, agent_id: str | None = None, query: str | None = None, schedule: ScheduleSpec | dict[str, Any] | None = None, enabled: bool = True, timeout_seconds: int | None = None, retry_policy: dict[str, Any] | None = None, overlap_policy: OverlapPolicy | str | None = None, session_policy: dict[str, Any] | None = None, workspace_policy: dict[str, Any] | None = None, metadata: dict[str, Any] | None = None, replace: bool = False) -> BackgroundTaskSpec: ...

Create a task: which agent runs which query, on what schedule (manual, once, interval or cron), with its timeout, retries, overlap, session and workspace policies. replace overwrites an existing one.

`resume_run`

async def resume_run(run_id: str) -> BackgroundRun: ...

Queue a run that was waiting for approval, once the approvals are decided (agent.resolve_approval). Its next attempt continues the same durable run instead of starting over.

`resume_task`

async def resume_task(task_id: str) -> None: ...

Schedule a paused task again.

`run_now`

async def run_now(task_id: str, query: str | None = None, wait: bool = False, timeout_seconds: float | None = None) -> BackgroundRun: ...

Queue one run of a task now, optionally with another query. With wait, return when it ends (or at timeout_seconds). Under the task’s overlap policy, a run can come back skipped (another run holds the task) instead of queued.

`run_until_terminal`

async def run_until_terminal(run_id: str, timeout_seconds: float | None = None, poll_interval_seconds: float = 0.05) -> BackgroundRun: ...

Execute a queued run in this process and return it when it ends, or at timeout_seconds.

`shutdown`

async def shutdown() -> None: ...

Stop the worker loop and cancel the runs it is executing; their attempts are recorded.

`start`

async def start() -> None: ...

Start the worker loop: it runs due schedules and queued runs until shutdown.

`unregister_agent`

async def unregister_agent(agent_id: str, force: bool = False) -> None: ...

Remove an agent. Refused while it has active runs, unless force.

`update_task`

async def update_task(task_id: str, patch: dict[str, Any]) -> BackgroundTaskSpec: ...

Change a task: patch holds the fields to change (query, schedule, enabled, timeout_seconds, retry_policy, overlap_policy, …).

`wait_for_run`

async def wait_for_run(run_id: str, timeout_seconds: float | None = None, poll_interval_seconds: float = 0.05) -> BackgroundRun: ...

Wait for one specific run to become terminal without executing work.

​BackgroundAgentManager

​Run statuses

​Overlap policies

​Methods

​cancel_run

​delete_task

​get_agent

​get_manager_status

​get_run

​get_run_events

​get_run_workspace

​get_task

​get_task_status

​initialize

​list_agents

​list_attempts

​list_runs

​list_tasks

​pause_task

​recover_expired_runs

​register_agent

​register_agent_spec

​register_task

​resume_run

​resume_task

​run_now

​run_until_terminal

​shutdown

​start

​unregister_agent

​update_task

​wait_for_run