REST API

The Dashboard REST API runs on port 9002 (listen_port + 2, configurable via dashboard_port in flo.toml). It powers the web console and can be used directly for monitoring, management, and ad-hoc operations.

Read endpoints serve projection data directly. Write endpoints are proposed through the node's normal replication path, so they are durable just like a CLI or SDK call.

Base URL

http://localhost:9002/api/v1

Authentication

GET /health is always public. The /api/v1/* endpoints require auth only when a key store exists (i.e. after flo server bootstrap); a dev node started without bootstrap serves them openly.

When auth is enabled:

POST /api/v1/auth/session     # body: { "api_key": "flo_sk_..." } → { "token": "..." }
GET  /api/v1/auth/status      # { "auth_enabled": true, ... }

Send the session token as a Bearer header on every request:

Authorization: Bearer <token>

Conventions

• Namespaces — Most resources are namespace-scoped. Pass ?namespace=<ns> (defaults to default).
• List vs. detail — A collection endpoint (e.g. GET /streams) returns an array of summaries; the detail endpoint (GET /streams/:name) returns the full record.

---

Cluster & metrics

GET /api/v1/cluster/stats

Cluster health, throughput, and the node/shard list.

{
  "rps": 45,
  "active_connections": 12,
  "uptime": "0d 4h 12m",
  "version": "0.12.0",
  "num_shards": 8,
  "commands_total": 1234567,
  "bytes_received": 9876543,
  "bytes_sent": 5432198,
  "subscriptions": 3,
  "nodes": [
    { "id": "shard-0", "status": "healthy", "role": "active" }
  ]
}

GET /api/v1/metrics

Aggregated metrics in JSON (Prometheus metrics are exposed separately on port 9001).

{
  "server": {
    "connections": 12,
    "subscriptions": 3,
    "commands_total": 1234567,
    "bytes_received": 9876543,
    "bytes_sent": 5432198,
    "uptime_seconds": 15120
  },
  "streams": 5,
  "queues": 3,
  "kv_namespaces": 4,
  "workflows": {
    "active_runs": 2, "started_total": 40, "completed_total": 31,
    "failed_total": 4, "cancelled_total": 1, "timed_out_total": 0,
    "signals_delivered_total": 6, "timers_fired_total": 0,
    "steps_executed_total": 120, "active_schedules": 1
  }
}

GET /health

Liveness and readiness check. Served at the root, not under /api/v1, and always public — it is never gated by the admin token, so it is safe to point an orchestrator probe at it without provisioning credentials.

{ "status": "ok" }

A 200 means the dashboard server is up and serving. Any other status, or a refused connection, means the node is not ready.

Stability: GET /health is a stable endpoint. The path, its public status, and the 200 + "status": "ok" contract will not change within a major version. Fields may be added to the object, so parse it leniently rather than matching the body exactly — checking the HTTP status alone is the most durable probe.

:::note This is served on the dashboard port (listen_port + 2, so 9002 by default), which requires [dashboard] enabled = true. [dashboard] bind defaults to 127.0.0.1, so a probe from outside the container needs bind = "0.0.0.0" — see Docker.

The Prometheus exporter serves its own separate /health on listen_port + 1, which additionally reports the shard count. :::

---

Namespaces

GET /api/v1/namespaces

[
  {
    "name": "production",
    "stream_count": 2, "queue_count": 1, "kv_count": 7,
    "workflow_count": 1, "processing_count": 3, "action_count": 3,
    "is_system": false
  }
]

POST /api/v1/namespaces

Create a namespace. Body: { "name": "production" }.

GET /api/v1/namespaces/:namespace

Namespace detail. Sub-collections: GET /namespaces/:namespace/streams, /queues, /kv.

---

KV

GET /api/v1/kv/namespaces

[
  { "name": "production", "key_count": 7, "bytes_stored": 1840,
    "get_ops": 0, "set_ops": 12, "delete_ops": 0 }
]

GET /api/v1/kv/namespaces/:namespace/keys?prefix=<p>&limit=<n>

Scan keys. Returns { keys: [{ key, size, version }], count, has_more, cursor, namespace } (here version is the entry LSN).

GET /api/v1/kv/namespaces/:namespace/keys/:key

{
  "found": true, "key": "user:42", "namespace": "production",
  "value": "...", "version": 3, "size": 128,
  "updated_at": 1718900000000, "ttl_ms": null
}

version here is the MVCC version count.

GET /api/v1/kv/namespaces/:namespace/keys/:key/history

{ "key": "user:42", "namespace": "production", "version_count": 3,
  "versions": [{ "version": 3, "term": 1, "timestamp_ms": 1718900000000, "size": 128, "tombstone": false }] }

PUT /api/v1/kv/namespaces/:namespace/keys/:key

Write a value. Body: { "value": "...", "ttl_seconds": 3600, "nx": false }.

DELETE /api/v1/kv/namespaces/:namespace/keys/:key

Delete a key.

---

Streams

GET /api/v1/streams?namespace=<ns>

[
  { "name": "events", "namespace": "production", "partitions": 16,
    "ingest_rate": 12840, "reads": 3110, "retention": "1d" }
]

ingest_rate and reads are cumulative record counts from the stream's metrics, not per-second rates — derive a rate by sampling across polls. retention is rendered from the stream's persisted policy ("1d", "1000 records", …), or "∞" when none is set.

GET /api/v1/streams/:name?namespace=<ns>

Detail with partitions[] (record counts + bytes) and consumer_groups[] (members, pending, last-delivered cursor).

GET /api/v1/streams/:name/messages?namespace=<ns>&limit=<n>

Records with id_ms, id_seq, ual_index, size, and the decoded payload. Sub-resources: GET /streams/:name/groups/:group, /groups/:group/pending, /groups/:group/members — all namespace-scoped via ?namespace=.

Writes

POST   /api/v1/streams/:name/trim?namespace=<ns>          # &max_len= | &max_age_s= | &max_bytes= | &dry_run=
DELETE /api/v1/streams/:name?namespace=<ns>&force=true    # force required for a non-empty stream
DELETE /api/v1/streams/:name/groups/:group?namespace=<ns>

trim requires at least one bound and returns { ok, stream, dry_run }. When several are supplied the most specific wins. Byte-based trim (max_bytes) is not implemented yet and returns an explicit error rather than silently doing nothing.

There is no stream compact endpoint: Flo streams are append-only and trim is their retention primitive.

---

Queues

GET /api/v1/queues

[
  { "name": "tasks", "namespace": "production", "ready": 24, "inflight": 0,
    "pending": 24, "available": 24, "enqueued": 30, "dequeued": 6, "dlq_count": 0 }
]

GET /api/v1/queues/:name?namespace=<ns>

Per-queue detail (same shape as the list item).

GET /api/v1/queues/:name/messages?namespace=<ns>&limit=<n>

Messages with seq, priority, state (ready / leased / dlq), attempts, lease_remaining_ms, size, and payload.

GET /api/v1/queues/:name/dlq?namespace=<ns>

Dead-letter entries.

Writes

POST   /api/v1/queues/:name?namespace=<ns>                # body: payload  (&priority= &delay_ms=)
POST   /api/v1/queues/:name/purge?namespace=<ns>
POST   /api/v1/queues/:name/dlq/:seq/requeue?namespace=<ns>
DELETE /api/v1/queues/:name/dlq/:seq?namespace=<ns>

Enqueue takes the raw request body as the message payload and returns { ok, queue, seq }. Purge removes every live (ready + leased) message and returns the real count as { ok, purged }; dead-letter entries are not per-queue addressable and are left intact.

---

Time Series

Measurements are namespace-scoped — pass ?namespace=.

GET /api/v1/timeseries?namespace=<ns>

[ { "name": "cpu_usage", "series_count": 1, "field_count": 1, "points": 60 } ]

GET /api/v1/timeseries/:measurement?namespace=<ns>

Detail: { name, namespace, field_count, fields: [{ name, type }], series_count, retention }.

GET /api/v1/timeseries/:measurement/data?field=<f>&namespace=<ns>&from=<ms>&to=<ms>

{ "measurement": "cpu_usage", "field": "value", "series": [ { "timestamp": 1718900000000, "value": 41.1 } ] }

Add &tags=host=web-01,env=prod to scope to matching tag-series. Tags are a real series dimension, and a filter constrains only the tags it names — so ?tags=host=web-01 also matches points tagged host=web-01,env=prod. Omit it to span every tag-series.

GET | POST /api/v1/timeseries/floql

Execute a FloQL query — ?q=<url-encoded> or a raw request body. The query is parsed, its source resolved across all shards, and the pipeline stages run server-side.

{
  "query": "cpu{host=web-01}[1h] | window(5m) | avg()",
  "series": [
    { "key": "cpu", "field": "value", "point_count": 12,
      "tags": [ { "key": "host", "value": "web-01" } ],
      "points": [ { "timestamp": 1718900000000, "value": 41.1 } ] }
  ]
}

Source tag filters are applied, including partial sets, !=, and the glob operators =~ / !~. A parse or execution failure returns the standard { "error": "…" } shape.

---

Actions

GET /api/v1/actions?namespace=<ns>

[
  { "name": "send-email", "namespace": "production", "type": "user",
    "owner": "platform-team", "version": 1, "enabled": true,
    "timeout_ms": 60000, "max_retries": 5, "worker_count": 1,
    "latency": { "count": 6, "avg_ms": 412, "p99_ms": 980 },
    "runs": { "total": 14, "pending": 7, "running": 0, "completed": 6,
              "failed": 1, "cancelled": 0, "timed_out": 0 } }
]

owner, timeout_ms and max_retries are persisted on the action record (set them with flo action register --owner --timeout --retries). latency is computed from completed runs — avg_ms over all of them, p99_ms over a bounded sample.

GET /api/v1/actions/:name?namespace=<ns>

Detail with runs, recent_runs[] (input/output/error/source), and the workers[] handling the action.

GET /api/v1/actions/:name/runs?namespace=<ns>&limit=<n>

Run history.

POST /api/v1/actions/:name/invoke?namespace=<ns>

Invoke (async). Body is the input JSON. Returns { ok, action, namespace, status, run_id }.

---

Workers

GET /api/v1/workers?namespace=<ns>

[
  { "worker_id": "worker-1", "status": "active", "worker_type": "action",
    "namespace": "production", "machine_id": null, "current_load": 0,
    "max_concurrent": 10, "tasks_completed": 6, "tasks_failed": 1,
    "last_seen": 1718900000000, "registered_at": 1718900000000, "metadata": null,
    "processes": [ { "name": "send-email", "kind": "action",
                     "run_count": 6, "fail_count": 1, "last_run_at": 1718900009000 } ] }
]

GET /api/v1/workers/:id

Single worker detail (same shape).

---

Processing

GET /api/v1/processing/jobs?namespace=<ns>

[
  { "job_id": "job-...", "name": "events-filter", "namespace": "production",
    "status": "RUNNING", "parallelism": 1, "batch_size": 100,
    "created_at": 1718900000000, "records_processed": 1000,
    "metrics": { "records_in": 1000, "records_out": 950,
                 "throughput_per_sec": 16.6, "output_per_sec": 15.8,
                 "last_read_count": 100, "batch_size": 100, "idle_ms": 210,
                 "watermark_ms": 1718899998000, "watermark_lag_ms": 2000,
                 "latency_last_ms": 1.4, "latency_avg_ms": 1.9,
                 "latency_max_ms": 7.2, "running": true } }
]

metrics is present on both the list and the detail. Rates are derived against the job's elapsed lifetime; last_read_count vs batch_size indicates saturation and idle_ms is the age of the last poll. running is false for a stopped job, where the live gauges read zero but cumulative counts remain. Latency is per-tick processing time, measured only on ticks that actually read records so idle polls don't skew the average.

GET /api/v1/processing/jobs/:id

Detail with the full pipeline yaml and savepoints[].

Writes

POST   /api/v1/processing/jobs?namespace=<ns>            # body: pipeline YAML → { ok, job_id, status }
PUT    /api/v1/processing/jobs/:id/stop?namespace=<ns>
DELETE /api/v1/processing/jobs/:id?namespace=<ns>        # cancel

---

Workflows

The console aliases workflows → workflow/definitions and workflows/:id → workflow/runs/:id.

GET /api/v1/workflow/definitions?namespace=<ns>

[
  { "name": "echo-workflow", "version": "1.0.0", "enabled": true,
    "step_count": 1, "plan_count": 0, "has_schedule": false,
    "has_trigger": false, "start_step": "start", "terminals": [], "steps": [] }
]

GET /api/v1/workflow/definitions/:name?namespace=<ns>

Detail with the full definition_yaml, status, and run_count.

PUT /api/v1/workflow/definitions/:name/enable?namespace=<ns>
PUT /api/v1/workflow/definitions/:name/disable?namespace=<ns>

GET /api/v1/workflow/runs?namespace=<ns>

[
  { "run_id": "wfr-...", "workflow": "echo-workflow", "version": "1.0.0",
    "status": "completed", "triggered_by": "manual", "current_step": null,
    "started_at": 1718900000000, "completed_at": 1718900000310,
    "duration_ms": 310, "error": null, "history_event_count": 6 }
]

POST /api/v1/workflow/runs?namespace=<ns>&workflow=<name>&version=<v>

Start a run. The request body is the input JSON. Returns { ok, run_id, workflow, status }.

GET /api/v1/workflow/runs/:id

Run detail with current_step, timing, input, output, step_results, and pending_signals.

GET /api/v1/workflow/runs/:id/history

Event timeline: [{ event_type, step_name, timestamp }].

POST   /api/v1/workflow/definitions?namespace=<ns>                    # body: workflow YAML/JSON
DELETE /api/v1/workflow/runs/:id?namespace=<ns>                       # cancel
POST   /api/v1/workflow/runs/:id/signal?namespace=<ns>&signal=<type>  # body: signal payload

POST /workflow/definitions defines a workflow from the request body and returns { ok, name, status }; the top-level name: in the definition is used to route it. signal requires ?signal=<type> (?type= is accepted too) — the request body is the optional payload.

---

Error Format

Errors are returned as a JSON object with an error field:

{ "error": "Human-readable description" }