Tasks & Execution Modes

Agentomatic ships a unified task subsystem that lets you run any platform resource — an agent, an ML plugin, a pipeline, a custom endpoint, or an ingestor — in synchronous, asynchronous (background), or batch modes, all through one consistent API. Every task also exposes a separate Server-Sent Events progress stream.

Every submission returns the same TaskRecord shape, so your frontend can submit work, poll status, subscribe to live progress, fetch the result, or cancel — regardless of what runs underneath.

Why one task API?

Instead of sprinkling bespoke async endpoints on every resource, the task board gives you a single, uniform contract for submit → track → result → cancel. It is the backbone for long-running work such as document ingestion, batch scoring, and multi-step pipelines.

Capabilities

Capability	Supported
Synchronous (blocking) execution	✅
Asynchronous background execution + task id	✅
Status polling (queued/running/succeeded/failed/cancelled)	✅
Live progress (percent / message / stage) over SSE	✅
Bounded in-process queue (concurrency limit)	✅
Batch fan-out with per-item progress	✅
Cancellation	✅
Completion webhooks (callback_url)	✅
A2A task lifecycle (real, pollable)	✅
Pluggable persistence (TaskStore)	✅

Per-resource execution modes

You don't have to use the task board directly. Every resource also gets ergonomic /async and /batch companions next to its synchronous route, all backed by the same task system. Submit to these and you get back a 202 with a task id and links to poll, stream, or cancel via the task board.

Resource	Sync	Async	Batch
Agent	POST /api/v1/{agent}/invoke	.../invoke/async	.../invoke/batch
Plugin	POST /api/v1/plugins/{name}/predict	.../predict/async	.../predict/batch
Pipeline	POST /api/v1/pipelines/{name}/run	.../run/async	.../run/batch
Ingestor	POST /api/v1/ingestion/{name}/run	.../run/async	.../run/batch
Endpoint	POST /api/v1/endpoints/{name}{path}	.../{path}/async	.../{path}/batch

The async body is identical to the sync body. The batch body wraps a list of those inputs:

# Fan out one agent over many inputs as a single batch task
curl -X POST http://localhost:8000/api/v1/researcher/invoke/batch \
  -H 'content-type: application/json' \
  -d '{"inputs": [{"query": "a"}, {"query": "b"}, {"query": "c"}],
       "callback_url": "https://example.com/webhook"}'
# -> 202 { "id": "task_...", "mode": "batch", "links": { ... } }

Poll the returned task id on the task board just like any other task; a batch task's result is the ordered list of per-item results.

The task board API

The task board is mounted at {api_prefix}/tasks (default /api/v1/tasks) and is enabled by default (enable_tasks=True).

Method	Path	Description
POST	/tasks	Submit work (202 async, 200 when wait=true)
GET	/tasks	List / filter tasks
GET	/tasks/{id}	Poll status + progress
GET	/tasks/{id}/result	Fetch the result (409 if still pending)
GET	/tasks/{id}/events	Resumable SSE progress stream
POST	/tasks/{id}/cancel	Request cancellation
DELETE	/tasks/{id}	Delete a terminal record

Submitting a task

Async (default)SynchronousBatch

curl -X POST http://localhost:8000/api/v1/tasks \
  -H 'content-type: application/json' \
  -d '{
    "target_type": "agent",
    "target": "researcher",
    "input": {"query": "What is Agentomatic?"}
  }'

Responds 202 immediately:

{
  "id": "task_a1b2c3d4e5f6a7b8",
  "target_type": "agent",
  "target": "researcher",
  "status": "queued",
  "progress": {"percent": null, "message": "", "current": 0, "total": null},
  "links": {
    "self": "tasks/task_a1b2c3d4e5f6a7b8",
    "result": "tasks/task_a1b2c3d4e5f6a7b8/result",
    "events": "tasks/task_a1b2c3d4e5f6a7b8/events",
    "cancel": "tasks/task_a1b2c3d4e5f6a7b8/cancel"
  }
}

Set wait: true (or mode: "sync") to block until the task is terminal. The endpoint returns 200 with the full record including result.

curl -X POST http://localhost:8000/api/v1/tasks \
  -H 'content-type: application/json' \
  -d '{
    "target_type": "plugin",
    "target": "sentiment_analyzer",
    "input": {"text": "I love this"},
    "wait": true,
    "timeout": 30
  }'

Provide batch (a list of inputs). Each item uses the selected resource's published JSON Schema and may therefore be an object, array, or scalar value. Items run with bounded concurrency and the task reports done/total progress. The result is a list aligned to the inputs (failed items become {"error": "..."}).

```bash
curl -X POST http://localhost:8000/api/v1/tasks \
  -H 'content-type: application/json' \
  -d '{
    "target_type": "agent",
    "target": "classifier",
    "batch": [{"query": "a"}, {"query": "b"}, {"query": "c"}]
  }'
```

With webhook

Provide callback_url to have the final record POSTed to your service on completion — no polling required.

curl -X POST http://localhost:8000/api/v1/tasks \
  -H 'content-type: application/json' \
  -d '{
    "target_type": "pipeline",
    "target": "ingest_docs",
    "input": {"source": "s3://bucket/docs/"},
    "callback_url": "https://myapp.example.com/hooks/task-done"
  }'

Resource input validation happens before queueing

POST /tasks applies the exact same published input contract as a resource's native run route. This includes a custom agent schema (or the standard invoke request), plugin prediction schemas, custom-endpoint schemas, and ingestor schemas. The same check is applied independently to every batch item. A malformed payload returns 422, and an unknown resource returns 404, before creating a durable task record. This prevents the Task Board from showing invalid work as queued or failed later.

Pipelines always require an object input. Their declared input_schema remains advisory by default, exactly as it is for POST /pipelines/{name}/run; set strict_schema: true to reject declared missing or mistyped inputs with 422 before queueing.

Use the resource's Studio schema panel or GET /studio/agents/{agent}/schemas to inspect the current deployed input and output contract before submitting work.

Polling status

curl http://localhost:8000/api/v1/tasks/task_a1b2c3d4e5f6a7b8

{
  "id": "task_a1b2c3d4e5f6a7b8",
  "status": "running",
  "progress": {"percent": 66.6, "message": "Completed 2/3 batch items", "current": 2, "total": 3},
  "duration_ms": 812.4
}

Streaming progress (SSE)

Subscribe to /tasks/{id}/events for live updates. Each frame is a TaskEvent; the stream ends with data: [DONE].

const es = new EventSource("/api/v1/tasks/task_a1b2c3d4e5f6a7b8/events");
es.onmessage = (e) => {
  if (e.data === "[DONE]") { es.close(); return; }
  const evt = JSON.parse(e.data);
  console.log(evt.sequence, evt.status, evt.progress?.percent);
};

Resuming a dropped stream

Every frame carries an SSE id: holding the event's sequence — a per-task counter that starts at 1 and never skips. That is what makes a dropped connection recoverable rather than merely survivable.

A browser EventSource needs no extra code: it reconnects on its own and sends the last id it saw back in Last-Event-ID, and the server replays only what was missed. Other clients pass the same value explicitly:

# Resume after event 7 — events 8 onwards, then live updates.
curl -N "http://localhost:8000/api/v1/tasks/task_a1b2c3d4e5f6a7b8/events?since=7"
curl -N -H "Last-Event-ID: 7" \
  http://localhost:8000/api/v1/tasks/task_a1b2c3d4e5f6a7b8/events

A client reconnecting without a cursor is treated as new: it receives the task's current snapshot (itself carrying an id:, so it can resume from there) rather than the whole history.

Replay is bounded, and says so

The event log keeps the last 512 events of each of the last 1024 tasks by default. If you resume from a point that has already been evicted, the stream sends a {"event": "truncated"} frame followed by the current snapshot — it never serves a partial history as if it were complete.

Tune it, or turn replay off, when constructing the manager:

from agentomatic.tasks import InMemoryTaskEventLog, TaskManager

manager = TaskManager(
    event_log=InMemoryTaskEventLog(max_events_per_task=2048, max_tasks=256)
)

One process, by default

The built-in event log lives in the worker that ran the task. Task records are shared through the store, so a client reconnecting to another worker still gets the correct snapshot and terminal result — but the missed intermediate events are not replayable from there.

For replay across workers or replicas, register a shared log. Agentomatic ships the interface, not a vendor client:

from agentomatic.tasks import TaskEventLog, register_event_log_provider

class RedisEventLog(TaskEventLog):
    async def append(self, event): ...
    async def replay(self, task_id, *, after=0): ...
    async def earliest_sequence(self, task_id): ...
    async def latest_sequence(self, task_id): ...
    async def drop(self, task_id): ...

register_event_log_provider("redis", lambda **cfg: RedisEventLog(**cfg))

This applies to every task target — agents, pipelines, plugins, endpoints and ingestion all run through the same manager, so a pipeline's per-step progress resumes exactly like an agent's.

Fetching the result

curl http://localhost:8000/api/v1/tasks/task_a1b2c3d4e5f6a7b8/result

• 200 with {"task_id": ..., "result": ...} when succeeded.
• 409 while the task is still pending.
• 422 if the task failed or was cancelled (the error is in detail).

Cancelling

curl -X POST http://localhost:8000/api/v1/tasks/task_a1b2c3d4e5f6a7b8/cancel

Running work is cancelled cooperatively; the record transitions to cancelled.

A2A task lifecycle

When the task subsystem is enabled, each agent's A2A endpoints run through the task manager and expose a real, pollable lifecycle (previously a stub):

Method	Path	Description
POST	/api/v1/{agent}/a2a/tasks	Submit an A2A task (async)
GET	/api/v1/{agent}/a2a/tasks/{id}	A2A task status
POST	/api/v1/{agent}/a2a/tasks/{id}/cancel	Cancel the task

Statuses are mapped to canonical A2A states: submitted, working, completed, failed, canceled.

Programmatic use

from agentomatic import TaskManager, TargetType
from agentomatic.tasks.dispatchers import make_agent_dispatcher

manager = TaskManager(max_concurrency=8)
manager.register_dispatcher(TargetType.AGENT, make_agent_dispatcher(registry))

record = await manager.submit(TargetType.AGENT, "researcher", input={"query": "hi"})
status = await manager.get(record.id)

# Or block until done:
done = await manager.submit_and_wait(TargetType.AGENT, "researcher", input={"query": "hi"})
print(done.status, done.result)

Reporting progress from your own code

Every dispatcher (and each ingestor's ingest) receives a TaskContext — use it to publish live progress (surfaced via /tasks/{id} and the SSE /events stream) and to honour cancellation:

from agentomatic.tasks import TaskContext

async def dispatcher(target: str, payload, ctx: TaskContext):
    total = len(payload["items"])
    for i, item in enumerate(payload["items"], start=1):
        if ctx.cancelled:            # cooperative cancellation
            break
        await ctx.report(percent=100 * i / total, message=f"item {i}/{total}",
                         current=i, total=total)
        ...
    return {"processed": total}

Nested graph nodes and pipeline step functions often do not receive the TaskContext argument. The platform installs a ContextVar bridge so you can report from anywhere under a task run:

from agentomatic.tasks import report_stage, report_stage_sync

async def my_node(state):
    await report_stage("embed", percent=40.0, message="Embedding documents")
    ...

def sync_node(state):
    report_stage_sync("encode", percent=20.0)  # schedules on the running loop
    ...

When no task is bound, both helpers are no-ops.

Configuration

from agentomatic import AgentPlatform

platform = AgentPlatform.from_folder(
    "agents/",
    enable_tasks=True,          # default
    task_max_concurrency=8,     # bounded in-process queue size
    # task_store=SQLAlchemyTaskStore(...),  # optional durable backend (see below)
)

TaskRecord

Field	Type	Description
id	str	Unique task id (task_...)
target_type	agent/plugin/pipeline/endpoint/ingestion	Resource kind
target	str	Resource name
mode	str	sync/async/batch
status	queued/running/succeeded/failed/cancelled	Lifecycle
progress	TaskProgress	percent, message, current, total, stage
input / batch	Any	Submitted payload(s)
result / error	Any / str	Terminal outcome
created_at / started_at / finished_at	float	Unix timestamps
duration_ms	float \| None	Wall-clock duration
callback_url	str \| None	Completion webhook

A task is failed when its target raises or reports failure: a pipeline whose status is failed, or an endpoint none of whose upstream calls succeeded. The target's own report is kept in result (every step's status and error, every upstream response) and summarised in error. A pipeline that finished with status partial (some steps failed under on_error: continue) is a completed run, so its task succeeded; check result.status.

Persistence & durability

The default InMemoryTaskStore keeps records in a bounded, TTL-aware dict — perfect for single-process deployments and tests. For multi-worker or restart-durable deployments, use the built-in SQLAlchemyTaskStore (or implement the TaskStore interface against any shared backend) and pass it via task_store=.

Durable SQLAlchemy store

SQLAlchemyTaskStore is a drop-in, durable TaskStore backed by any SQLAlchemy async driver. Records survive process restarts and are shared across workers/replicas through a common database.

from agentomatic import AgentPlatform
from agentomatic.tasks import SQLAlchemyTaskStore

# SQLite for development…
store = SQLAlchemyTaskStore("sqlite+aiosqlite:///data/tasks.db")

# …PostgreSQL for production.
store = SQLAlchemyTaskStore("postgresql+asyncpg://user:pass@localhost/db")

platform = AgentPlatform.from_folder("agents/", task_store=store)

Install the optional dependency:

SQLitePostgreSQL

pip install "agentomatic[db]"

pip install "agentomatic[db-postgres]"

The platform initialises and disposes the store automatically as part of its lifespan — no manual initialize()/close() needed.

Configuration (all optional, safe defaults):

Argument	Default	Description
url	sqlite+aiosqlite:///data/tasks.db	Any SQLAlchemy async URL
table_name	agentomatic_tasks	Table to store records in
pool_size / max_overflow / pool_recycle / pool_pre_ping	10 / 20 / 3600 / True	Connection pool (ignored for SQLite)
echo	False	Log all SQL (debug)
ttl_seconds	604800 (7 days)	Terminal records older than this are evicted; None disables
max_records	100000	Soft cap; oldest terminal records evicted first; None disables
eviction_interval	200	Run eviction at most once every N saves
engine	None	Reuse an existing async engine (e.g. an agent's DatabaseConnection); not disposed on close()

Forward-compatible schema

Records are stored as an indexed JSON payload plus a handful of indexed scalar columns (status, target, created_at, …), so adding fields to TaskRecord never requires a migration. Eviction is best-effort and never blocks or fails a save.

Importing is always safe

agentomatic.tasks never imports SQLAlchemy eagerly. The dependency is only required when you actually construct SQLAlchemyTaskStore; a clear install hint is raised otherwise.

Multi-worker deployments

The in-memory store and in-process queue are per-process. If you run with workers > 1, provide a shared TaskStore (the SQLAlchemyTaskStore works out of the box) so status and results are consistent across workers.