REST, SSE and MCP

The engine (pipod) serves one HTTP server, the gateway, with five parts:

The gateway runs when engine.listen is set in config.yaml, or the engine is started with --listen:

pipo engine start --listen 8787     # 0 picks a free port
pipo engine status                  # prints the port; it is also in <home>/run/engine.json

It binds to 127.0.0.1 unless engine.host says otherwise (see Owner and remote access). Every CLI command has --json, which gives the same data as this API.

Conventions#

The owner only. With engine.auth: owner (the default), /api and /events need the owner's session cookie (from /api/auth/login) or the engine key, Authorization: Bearer <contents of <home>/engine.key>, which the CLI sends. Without either they answer 401 with code login_required, or owner_setup_required before the owner exists. /api/auth/* needs neither. See Owner and remote access.

Addressed to this engine. /api, /events, /mcp and the OAuth routes answer only requests whose Host is a loopback name (127.0.0.1, localhost, [::1]) or the host of engine.public_url, and refuse an Origin other than the engine's own. Anything else is 403. Every POST to /api must have Content-Type: application/json (415 otherwise). Together, these stop a web page in your browser from driving the engine (DNS rebinding, cross-site form posts). /in has no such limits: inputs have their own auth.

Errors are JSON with a status that matches:

{ "error": "no pipeline named 'people-intak' in this engine", "hint": "GET /api/pipelines lists them; start one with POST /api/pipelines {file} (pipo start <file>)", "code": "not_found" }

A diagnostics list has the shape of pipo check --json: code, severity, message, hint, line, col, path, file.

Names and ids. Pipeline names are [a-z0-9-]. Versions may be given as 3 or v3.

Reads without a runner. Packet, DLQ, version and proposal reads ask the runner when it runs. Otherwise they read the pipeline's journal read-only. The answer's source says which: runner or journal. A payload that can't be redacted safely without the runner (a secret that can't be resolved here) is withheld and listed in withheld.

Writes need a runner. DLQ replay and purge, rerun, rollback, proposals and resolve are committed by the runner. With the pipeline stopped they answer 409 with a hint to start it.

Owner#

Engine#

code_changed: true means Pipo's code on disk changed since this engine started; the dashboard then offers a restart. GET /api/engine answers even while the engine is starting.

Pipelines#

Each entry of GET /api/pipelines:

Start a pipeline and wait for it:

curl -s localhost:8787/api/pipelines -H 'content-type: application/json' \
  -d "{\"file\": \"$PWD/people-intake.pipo\"}"
curl -s localhost:8787/api/pipelines/people-intake | jq '.state, .runner.stats'

A start that fails pipo check answers 422 with the diagnostics. A taken port answers 409 and names the process holding it.

Packets#

curl -s localhost:8787/api/pipelines/people-intake/push -H 'content-type: application/json' \
  -d '{"data": {"name": "Ada", "age": 36}, "source": "manual"}'
curl -s 'localhost:8787/api/pipelines/people-intake/packets?state=dead_lettered&limit=10'

Dead-letter queue#

Give ids (a list) or all: true.

Versions and proposals#

See Live changes and proposals and Versions and rollback.

Events#

GET /events: the live stream#

Server-Sent Events of every journal event of every pipeline, plus pipeline state changes:

curl -N localhost:8787/events?pipeline=people-intake

retry: 1000

id: people-intake:1042
data: {"id":"people-intake:1042","pipeline":"people-intake","seq":1042,"at":1760083200000,"packet_id":"01J9Z…","type":"packet.delivered","node":null,"detail":{…}}

event: state
data: {"name":"people-intake","state":"running","status":"paused",…}

• Each journal event is a data: line with id: <pipeline>:<seq>. The object has pipeline, seq (increasing per pipeline), at (epoch ms), packet_id, type (such as packet.accepted, packet.delivered, pipeline.paused), node and detail.
• event: state carries a pipeline's new supervision state (the same object as an entry of GET /api/pipelines). It has no id.
• A comment is sent every 15 s to keep the connection open.
• ?pipeline=<name> limits the stream to one pipeline.

Reconnecting. Browsers' EventSource sends Last-Event-ID by itself; other clients can pass ?after=<id>. With ?pipeline=, the stream replays exactly the journal's events after that seq (up to 10 000, then an event: gap with a hint), then continues live without repeats. Without ?pipeline=, it replays from the last 1000 events the engine streamed; when the id is older, or the engine restarted, it sends event: gap with a hint. A client more than 16 MiB behind is dropped and replays when it reconnects.

An open stream keeps the engine awake: it doesn't go to sleep while a dashboard is open.

GET /api/events: a page#

GET /api/events?pipeline=&after=&limit= returns {events, more, last_id}: the last limit events (default 100, 1–1000), or those after after. Page by passing last_id back as after. With pipeline, it reads that pipeline's journal, so it works for any seq. Without it, after must still be among the engine's recent events (410 otherwise).

Ingress: /in#

Webhooks reach a pipeline's http input at /in/<pipeline><path>, where <path> is the input's with.path:

curl -X POST localhost:8787/in/people-intake/people \
  -H 'X-Pipo-Token: dev-token' -H 'content-type: application/json' -d '{"name": "Ada", "age": 36}'

The gateway forwards the request unchanged (method, path, query, raw body and headers, plus X-Forwarded-For, -Host and -Proto) to the runner's own port, and passes the runner's answer back as is. The runner answers:

A csv body can hold several records; the answer is then {"packets": [...]}, one entry per record.

The gateway adds its own errors. It never retries:

The same route is served directly on the runner's own port (with.listen, or --listen), which keeps working while the engine is down. pipo run --listen <port> serves it without any engine. See Running pipelines.

Builder#

The dashboard's builder uses these routes. They read and write .pipo files, but never start, stop or change a running pipeline.

Files must be absolute paths to .pipo files inside the workspace, or the file of a pipeline the engine knows. Anything else is 403. Saving an existing file needs overwrite: true (409 otherwise), and sources are capped at 1 MiB (413).

Bots and settings#

See Telegram bots.

Agent approvals#

Pipelines agents wrote with create_pipeline, and a human's decision on them (see Agents that build pipelines).

Starting a pipeline any other way as a human (POST /api/pipelines, …/start, …/restart) approves it too.

MCP: /mcp#

The MCP endpoint takes one JSON-RPC message per POST, authenticated with Authorization: Bearer <token> (an OAuth access token, or a static token). Its tools call the REST handlers above, so the answers match. Tokens, scopes, tools and resources are described in Agents as operators.

curl -s localhost:8787/mcp -H "Authorization: Bearer $PIPO_MCP_TOKEN" -H 'content-type: application/json' \
  -d '{"jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": {"name": "get_status", "arguments": {"pipeline": "people-intake"}}}'

OAuth#

The MCP endpoint's OAuth server (see Agents as operators). Its errors are OAuth's {error, error_description}, not the /api shape.

The consent page and the grant list use /api:

The runner's control socket#

Underneath the engine, each runner listens on a Unix socket, <home>/run/<name>.sock, with newline-delimited JSON requests {id, op, args}. The engine, the CLI's --no-engine mode and chains use it. It is an internal protocol: use REST or the CLI instead. The spec documents it in §7.2.

See also#

• The engine
• Configuration
• CLI reference