Orchestrate your pipelines.
Real work has steps that depend on each other — some can run in parallel, some have to wait, some fail and need a retry. Annum runs the whole graph for you: it works out the order, runs what it can at once, tracks every step, and reruns what breaks.
The idea
What a workflow engine does for you
Most real tasks aren't a single step. You fetch something, transform it, score it, summarize it — and each step needs the one before it. Some steps could run side by side; others have to wait their turn. When one fails, you want to retry just that step. And often you need to run the same process over hundreds of inputs at once.
Wiring all of that up by hand — the ordering, the parallelism, the retries, the bookkeeping — quietly becomes its own project. That's what a workflow engine is for. You describe your steps and how they connect as a graph, and the engine takes it from there: it figures out what's ready, runs independent steps at the same time, remembers what's done, reruns what breaks, and shows live progress the whole way.
Annum is that engine — small and API-first. You hand it the graph; your own workers do the actual work.
In practice
Describe the graph. Point your workers at it.
A template is plain JSON: the steps, and what each one depends on. Your workers pick up tasks by type, do the work, and report back — so the same template can fan out over hundreds of inputs at once.
# 3 sources → select a series → 3 projections → report
{
"name": "dcf-valuation",
"dag": { "nodes": [
{ "name": "filings", "task_type": "get_filings" },
{ "name": "prices", "task_type": "get_prices" },
{ "name": "rates", "task_type": "get_rates" },
{ "name": "series", "task_type": "select", "deps": ["filings", "prices", "rates"] },
{ "name": "base", "task_type": "project", "params": {"scenario": "base"}, "deps": ["series"] },
{ "name": "bull", "task_type": "project", "params": {"scenario": "bull"}, "deps": ["series"] },
{ "name": "bear", "task_type": "project", "params": {"scenario": "bear"}, "deps": ["series"] },
{ "name": "report", "task_type": "dcf_report", "deps": ["base", "bull", "bear"] }
] }
}from annum_worker import Worker
w = Worker("https://annum.aisloppy.com", api_key=KEY)
w.register("get_filings", get_filings) # fn(inputs, params)
w.register("get_prices", get_prices)
w.register("get_rates", get_rates)
w.register("select", select_series)
w.register("project", project) # one handler, every scenario
w.register("dcf_report", dcf_report)
w.run() # claim → heartbeat → complete/fail loop
# Value many tickers at once → one run each
# POST /api/runs:batch
{ "dag_template_id": 1, "entities": ["AAPL", "MSFT", "NVDA"] }Ship your first pipeline
Sign in to watch live runs, or grab a key and start submitting dag_templates from your own services.