Connectors

Every built-in connector and its with: settings, generated from the connector manifests that pipo check, the JSON Schema and the builder's forms all use.

Every block follows the same pattern: verb: connector, then with: for that connector's settings. pipo check validates each with: against the schema below; unknown keys are errors (P002).

Inputs: http, schedule, watch, push, pipeline, system, telegram, poll, tail, command, crawl
Taps: log, http, file, telegram, emit, exec
Transforms: map, http, exec
Agent providers: claude_api, claude_code, codex, pi, opencode
Outputs: sqlite, file, http, telegram, pipeline, stdout
Delivery checks: ack, external, none, record_exists, row_count, query, file_exists, file_nonempty, line_contains, checksum, status, downstream, follow_up

Inputs#

Set with input.via (or inputs.<name>.via). Guide: Inputs.

via: http#

Webhook or API call to /in/<pipeline><path>.

Field	Type	Meaning
`path`	string	Route below `/in/<pipeline>`; default `/`. (default `"/"`)
`method`	`GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE`	HTTP method accepted; default `POST`.
`auth`	object	Request authentication: `header` + `equals` (a token header), or `hmac: {header, secret, algorithm}` (a body signature; `sha256` by default).
`public`	boolean	`true` takes requests from the network without `auth` when the engine is exposed (`engine.host`); without it such requests get `403`. Loopback callers never need it.
`respond`	`accepted` \| `delivered`	`accepted` (default): answer `202 {packet_id}` once journaled; `delivered`: wait for the final state up to `timeout`.
`timeout`	duration	How long `respond: delivered` waits for delivery; default `30s`.
`listen`	integer	A port the runner binds itself on 127.0.0.1, so the input works without the engine gateway. (1–65535)

via: schedule#

Timer, by cron expression or fixed interval.

Field	Type	Meaning
`cron`	string	5-field cron expression, evaluated in UTC. Exactly one of `cron` and `every` (P038).
`every`	duration	Fixed interval, counted from when the runner starts.
`payload`	any	The `data` of every tick; default `{}`.

via: watch#

File system changes.

Field	Type	Required	Meaning
`path`	string	yes	Glob of files to watch, relative to the pipeline file; a plain folder means every file below it.
`events`	list of `create`, `change`, `delete`		Which changes become packets; default `[create, change]`.
`read`	`content` \| `path`		`path` (default): describe the file; `content`: also read its text into `data.content`.

Example data of one packet:

{
  "event": "create",
  "path": "/home/me/inbox/note.txt",
  "name": "note.txt",
  "size": 12,
  "mtime": "2026-01-01T09:00:00.000Z",
  "content": "hello there\n"
}

via: push#

Packets pushed by pipo push, the UI or an agent.

No with: settings.

via: pipeline#

Packets from another pipeline's to: pipeline output.

Field	Type	Required	Meaning
`from`	list of string	yes	The pipelines allowed to feed this input with their `to: pipeline` output; others are refused.

via: system#

Operating system samples.

Field	Type	Required	Meaning
`every`	duration	yes	How often to sample.
`metrics`	list of `cpu`, `memory`, `disk`, `battery`, `network`		Which metrics to sample; default all of them.

Example data of one packet:

{
  "cpu": {
    "percent": 12.5,
    "cores": 8,
    "load": [
      0.42,
      0.38,
      0.3
    ]
  },
  "memory": {
    "total": 17179869184,
    "free": 4294967296,
    "used": 12884901888,
    "percent": 75
  },
  "disk": {
    "path": "/",
    "total": 499963174912,
    "free": 199985269964,
    "used": 299977904948,
    "percent": 60
  },
  "battery": {
    "percent": 80,
    "status": "Discharging"
  },
  "network": {
    "rx_bytes": 123456789,
    "tx_bytes": 9876543
  },
  "sampled_at": "2026-01-01T09:00:00.000Z"
}

via: telegram#

Messages to a Telegram bot (long polling while the pipeline runs).

Field	Type	Meaning
`bot`	string	A bot name from `<home>/bots.json` (never a template); default the home's default bot.
`token`	string	A bot token instead of a named bot, e.g. `${secrets.tg_token}`.
`allow`	list of integer	Chat or user ids whose messages become packets; replaces the bot's own list. Messages from anyone else are ignored.
`poll_every`	duration	How long one long-poll waits for messages; default the bot's, else `25s`, at least `1s`.
`download`	boolean	Download attached files into the pipeline's folder; default `true`.

Example data of one packet:

{
  "message_id": 42,
  "date": "2026-01-01T09:00:00.000Z",
  "chat_id": 123456789,
  "chat_type": "private",
  "from": {
    "id": 123456789,
    "username": "ada",
    "name": "Ada Lovelace"
  },
  "text": "hello",
  "file": {
    "kind": "document",
    "file_id": "BQACAgQAAxkBAAI",
    "name": "report.pdf",
    "mime_type": "application/pdf",
    "size": 52000,
    "path": "/home/me/.pipo/pipelines/bot/files/report.pdf"
  }
}

via: poll#

Fetch a URL on a timer and emit what is new or changed.

Field	Type	Required	Meaning
`url`	string	yes	The URL to fetch (http or https), a template rendered for each fetch with `env`, `secrets` and `cursor`.
`cron`	string		5-field cron expression, evaluated in UTC. Exactly one of `cron` and `every` (P038).
`every`	duration		Fetch at start, then this long after each fetch; a restart waits out the rest.
`method`	`GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE`		HTTP method; default `GET`.
`headers`	map		Request headers, templates rendered for each fetch (`env`, `secrets`, `cursor`).
`body`	any		Request body for methods other than GET and DELETE: a string as text, anything else as JSON.
`timeout`	duration		How long one request may take; default `30s`.
`success`	list of integer or string		Status codes that count as success, as `204`, `"2xx"` or `"200-299"`; default `2xx`.
`parse`	`auto` \| `json` \| `text` \| `feed` \| `html`		How the answer becomes `data`: `auto` (default, by content type), `json`, `text`, `feed` (RSS, Atom, JSON Feed) or `html`.
`select`	string		A CSS selector (`parse: html`): `text` and `html` hold only the matching elements.
`items`	string		Expression over `data`, `response` and `cursor` giving the list to split into packets; default the whole answer (a feed's entries).
`key`	string		Expression over `item` and `data` naming an element; default a feed entry's `id`, the URL for a whole answer, else a hash.
`emit`	`new` \| `changed` \| `all`		`new`: unseen keys; `changed`: also changed content; `all`: everything. Default `changed` for a whole answer, else `new`.
`backfill`	boolean		`false`: take what the first fetch finds as seen and send only what comes later; default `true`.
`remember`	integer		How many keys are kept; keys that stopped appearing are forgotten first. Default `10000`. (≥ 1)
`cursor`	string		Expression over `data`, `items`, `response` and `cursor`, saved after each fetch and available as `${cursor}` in the next request.
`next`	string		Expression over `data` and `response` giving the next page's URL, followed in the same fetch.
`max_pages`	integer		The most pages one fetch follows through `next`; default `10`. (1–1000)

via: tail#

Lines appended to a file.

Field	Type	Required	Meaning
`path`	string	yes	The file to follow, relative to the pipeline file.
`parse`	`text` \| `json`		`text` (default): `{line, path}`; `json`: the line parsed as JSON (others are skipped).
`from`	`end` \| `start`		Where a first start begins: `end` (default, only new lines) or `start` (the whole file).

Example data of one packet:

{
  "line": "127.0.0.1 - - [01/Jan/2026:09:00:00 +0000] \"GET / HTTP/1.1\" 200 512",
  "path": "/var/log/app.log"
}

via: command#

Lines a long-running program prints.

Field	Type	Required	Meaning
`command`	string	yes	The program to run (no shell), on PATH or a path relative to the pipeline file.
`args`	list of string		Arguments, one string each, passed as they are.
`cwd`	string		Where the program runs, relative to the pipeline file; default its folder.
`env`	map		Environment variables added for the program.
`parse`	`text` \| `json`		`text` (default): `{line}`; `json`: each stdout line parsed as JSON (others are skipped).
`restart`	`always` \| `on-failure` \| `never`		When the program exits: `always` (default), `on-failure` (an error exit or a signal) or `never`.

Example data of one packet:

{
  "line": "Jan 01 09:00:00 host nginx[42]: started"
}

via: crawl#

Crawl a site on a timer and emit the pages that changed.

Field	Type	Required	Meaning
`start`	any	yes	The URL, or list of URLs, each crawl starts from.
`cron`	string		5-field cron expression, evaluated in UTC. Exactly one of `cron` and `every` (P038).
`every`	duration		Crawl at start, then this long after each crawl ended.
`within`	any		URL prefixes a link must start with to be followed; default each start URL's origin.
`exclude`	list of string		Regular expressions; a URL matching one is not followed.
`depth`	integer		How many links deep to follow from `start` (depth 0); default `3`. (0–100)
`max_pages`	integer		The most pages one crawl fetches; default `500`. (1–100000)
`delay`	duration		The wait between requests; default `1s` (a `Crawl-delay` in robots.txt can raise it).
`select`	string		A CSS selector: each page's `text` and `html` hold only the matching elements.
`headers`	map		Headers sent with every request (templates rendered at start).
`timeout`	duration		How long one request may take; default `30s`.
`emit`	`changed` \| `all`		`changed` (default): pages that are new or changed since the last crawl; `all`: every page.
`remember`	integer		How many pages are kept; pages that stopped appearing are forgotten first. Default `10000`. (≥ 1)
`robots`	boolean		Obey robots.txt; default `true`. Turn it off only for your own sites.

Example data of one packet:

{
  "url": "https://docs.example.com/guide/intro",
  "status": 200,
  "title": "Introduction",
  "text": "Welcome to the guide.",
  "links": [
    "https://docs.example.com/guide/install"
  ],
  "html": null,
  "depth": 1
}

Taps#

Side effects in the middle of the graph; data passes on unchanged. tap: fn.<name> calls a user function. Guide: Nodes and the graph.

tap: log#

Write a log line.

Field	Type	Required	Meaning
`level`	`debug` \| `info` \| `warn` \| `error`		Log level; default `info`.
`message`	string		The line to log (a template); default the packet's `data` as JSON.

tap: http#

Call an HTTP endpoint; data passes on unchanged.

Field	Type	Required	Meaning
`method`	`GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE`		HTTP method; default `POST`.
`url`	string	yes	The URL to call (a template).
`headers`	map		Request headers (templates). An `Idempotency-Key: <key>` header is sent unless you set one.
`body`	any		Request body (a template); default the packet's `data`. A string is sent as text, anything else as JSON. Not sent for GET and DELETE.
`success`	list of integer or string		Status codes that count as success, as `204`, `"2xx"` or `"200-299"`; default `2xx`.

tap: file#

Append or write to a file.

Field	Type	Required	Meaning
`path`	string	yes	File to write, relative to the pipeline file (a template).
`format`	`jsonl` \| `json` \| `csv` \| `text`		How each packet is written; default `jsonl`.
`mode`	`append` \| `write`		`append` (default) or `write` (replace the file).

tap: telegram#

Send a Telegram message; data passes on unchanged.

Field	Type	Meaning
`bot`	string	A bot name from `<home>/bots.json` (never a template); default the home's default bot.
`token`	string	A bot token instead of a named bot, e.g. `${secrets.tg_token}`.
`chat_id`	integer or string	Chat to send to; default the chat a telegram input's packet came from (required with any other input, P057).
`text`	string	Message text, or a file's caption; default `data` (as JSON when not a string), cut at 4096 characters.
`photo`	string	A photo to send: a path relative to the pipeline file, or an http(s) URL.
`document`	string	A file to send: a path relative to the pipeline file, or an http(s) URL.
`parse_mode`	`Markdown` \| `MarkdownV2` \| `HTML`	Telegram formatting of `text`; off by default, so text from data can't break it.

tap: emit#

Emit a custom event.

Field	Type	Required	Meaning
`event`	string	yes	Name of the custom event to record in the journal and the event stream.
`detail`	any		Data attached to the event (a template).

tap: exec#

Run a program on this machine; data passes on unchanged.

Field	Type	Required	Meaning
`command`	string	yes	The program: a name found on `PATH`, or a path relative to the pipeline file. No shell.
`args`	list of string		Arguments, each passed to the program as one argument (templates).
`cwd`	string		Folder to run in, relative to the pipeline file; default the pipeline file's folder.
`env`	map		Extra environment variables for the program.
`stdin`	string		Text fed to the program's standard input.
`timeout`	duration		Kill the program after this long; default `5m`.
`success`	list of integer		Exit codes that count as success; default `[0]`.
`result`	`info` \| `text` \| `json`		What a transform's `data` becomes: `info` (default; exit code, stdout, stderr, duration, files), `text` (stdout) or `json` (stdout parsed).
`outputs`	list of string		Files the program must write: their folders are created first, and a missing one is a node error.

Transforms#

Replace data with a result. transform: fn.<name> calls a user function. Guide: Nodes and the graph.

transform: map#

Replace data with the rendered with.data template.

Field	Type	Required	Meaning
`data`	any	yes	The new `data`: a template, rendered per packet (a whole `${expr}` keeps its type).

transform: http#

Call an HTTP endpoint; the response body becomes data.

Field	Type	Required	Meaning
`method`	`GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE`		HTTP method; default `POST`.
`url`	string	yes	The URL to call (a template).
`headers`	map		Request headers (templates). An `Idempotency-Key: <key>` header is sent unless you set one.
`body`	any		Request body (a template); default the packet's `data`. A string is sent as text, anything else as JSON. Not sent for GET and DELETE.
`success`	list of integer or string		Status codes that count as success, as `204`, `"2xx"` or `"200-299"`; default `2xx`.

transform: exec#

Run a program on this machine; its result becomes data.

Field	Type	Required	Meaning
`command`	string	yes	The program: a name found on `PATH`, or a path relative to the pipeline file. No shell.
`args`	list of string		Arguments, each passed to the program as one argument (templates).
`cwd`	string		Folder to run in, relative to the pipeline file; default the pipeline file's folder.
`env`	map		Extra environment variables for the program.
`stdin`	string		Text fed to the program's standard input.
`timeout`	duration		Kill the program after this long; default `5m`.
`success`	list of integer		Exit codes that count as success; default `[0]`.
`result`	`info` \| `text` \| `json`		What a transform's `data` becomes: `info` (default; exit code, stdout, stderr, duration, files), `text` (stdout) or `json` (stdout parsed).
`outputs`	list of string		Files the program must write: their folders are created first, and a missing one is a node error.

Agent providers#

An agent: node's provider. Guide: Agent nodes.

agent: claude_api#

Claude via the Anthropic API (needs an API key).

Runs: a model API · default timeout: 60s

Field	Type	Required	Meaning
`model`	string	yes	Model id, e.g. `claude-sonnet-5-5`.
`prompt`	string	yes	The prompt (a template); use `${json(data)}` to include the packet.
`schema`	string	yes	JSON Schema file the answer must match, relative to the pipeline file.
`timeout`	duration		How long one call may take; default `60s`.
`max_tokens`	integer		Most output tokens per call; default 4096 (and never more than the packet's budget has left). (≥ 1)

agent: claude_code#

Your local Claude Code CLI (claude), logged in as you.

Runs: a local CLI (claude) · default timeout: 5m

Field	Type	Required	Meaning
`model`	string		Model to ask the CLI for; default the CLI's own.
`prompt`	string	yes	The prompt (a template); use `${json(data)}` to include the packet.
`schema`	string	yes	JSON Schema file the answer must match, relative to the pipeline file.
`timeout`	duration		How long one call may take; default `5m`.
`cwd`	string		Folder to run in, relative to the pipeline file; default a fresh, empty folder per call.
`allow_tools`	boolean		Let the CLI use its tools (edit files, run commands) without asking; default `false`.

agent: codex#

Your local OpenAI Codex CLI (codex), logged in as you.

Runs: a local CLI (codex) · default timeout: 5m

Field	Type	Required	Meaning
`model`	string		Model to ask the CLI for; default the CLI's own.
`prompt`	string	yes	The prompt (a template); use `${json(data)}` to include the packet.
`schema`	string	yes	JSON Schema file the answer must match, relative to the pipeline file.
`timeout`	duration		How long one call may take; default `5m`.
`cwd`	string		Folder to run in, relative to the pipeline file; default a fresh, empty folder per call.
`allow_tools`	boolean		Let the CLI use its tools (edit files, run commands) without asking; default `false`.

agent: pi#

Your local pi coding agent CLI (pi), with its configured providers.

Runs: a local CLI (pi) · default timeout: 5m

Field	Type	Required	Meaning
`model`	string		Model to ask the CLI for; default the CLI's own.
`prompt`	string	yes	The prompt (a template); use `${json(data)}` to include the packet.
`schema`	string	yes	JSON Schema file the answer must match, relative to the pipeline file.
`timeout`	duration		How long one call may take; default `5m`.
`cwd`	string		Folder to run in, relative to the pipeline file; default a fresh, empty folder per call.
`allow_tools`	boolean		Let the CLI use its tools (edit files, run commands) without asking; default `false`.

agent: opencode#

Your local opencode CLI (opencode), with its configured providers.

Runs: a local CLI (opencode) · default timeout: 5m

Field	Type	Required	Meaning
`model`	string		Model to ask the CLI for; default the CLI's own.
`prompt`	string	yes	The prompt (a template); use `${json(data)}` to include the packet.
`schema`	string	yes	JSON Schema file the answer must match, relative to the pipeline file.
`timeout`	duration		How long one call may take; default `5m`.
`cwd`	string		Folder to run in, relative to the pipeline file; default a fresh, empty folder per call.
`allow_tools`	boolean		Let the CLI use its tools (edit files, run commands) without asking; default `false`.

Outputs#

Set with output.to. Every pipeline has exactly one output. Guide: Outputs.

to: sqlite#

SQLite table.

Delivery checks: ack, external, none, record_exists, row_count, query · Batching: yes

Field	Type	Required	Meaning
`path`	string	yes	Database file, relative to the pipeline file.
`table`	string	yes	Table to write to.
`create`	boolean		Create the table (with the key as PRIMARY KEY) when it's missing; default `false`.
`mode`	`insert` \| `upsert`		`insert` (default): a repeated key is ignored; `upsert`: it updates the row.
`key`	string		The idempotency key column; default `packet_id`. It needs a PRIMARY KEY or UNIQUE constraint.
`columns`	map		Column → template map; default the top-level fields of `data` plus the key column set to the packet's key.

to: file#

File on disk.

Delivery checks: ack, external, none, file_exists, file_nonempty, line_contains, checksum · Batching: yes

Field	Type	Required	Meaning
`path`	string	yes	File to write, relative to the pipeline file (a template).
`format`	`jsonl` \| `json` \| `csv` \| `text`		`jsonl` (default; each line carries `packet_id`, so a repeated write is skipped), `json`, `csv` or `text`.
`mode`	`append` \| `write`		`append` (default) or `write` (replace the file).

to: http#

HTTP endpoint.

Delivery checks: ack, external, none, status, follow_up · Batching: no

Field	Type	Required	Meaning
`method`	`GET` \| `POST` \| `PUT` \| `PATCH` \| `DELETE`		HTTP method; default `POST`.
`url`	string	yes	The URL to call (a template).
`headers`	map		Request headers (templates). An `Idempotency-Key: <key>` header is sent unless you set one.
`body`	any		Request body (a template); default the packet's `data`. A string is sent as text, anything else as JSON. Not sent for GET and DELETE.
`success`	list of integer or string		Status codes that count as success, as `204`, `"2xx"` or `"200-299"`; default `2xx`.

to: telegram#

Telegram message.

Delivery checks: ack, external, none · Batching: no

Field	Type	Meaning
`bot`	string	A bot name from `<home>/bots.json` (never a template); default the home's default bot.
`token`	string	A bot token instead of a named bot, e.g. `${secrets.tg_token}`.
`chat_id`	integer or string	Chat to send to; default the chat a telegram input's packet came from (required with any other input, P057).
`text`	string	Message text, or a file's caption; default `data` (as JSON when not a string), cut at 4096 characters.
`photo`	string	A photo to send: a path relative to the pipeline file, or an http(s) URL.
`document`	string	A file to send: a path relative to the pipeline file, or an http(s) URL.
`parse_mode`	`Markdown` \| `MarkdownV2` \| `HTML`	Telegram formatting of `text`; off by default, so text from data can't break it.

to: pipeline#

Another pipeline, fed through its via: pipeline input (§3.14).

Delivery checks: ack, external, none, downstream · Batching: no

Field	Type	Required	Meaning
`pipeline`	string	yes	The pipeline to feed; its `via: pipeline` input must list this one in `from`.
`data`	any		What to hand over (a template); default the packet's `data`.

to: stdout#

Standard output.

Delivery checks: ack, external, none · Batching: no

Field	Type	Required	Meaning
`format`	`jsonl` \| `json` \| `text`		How each packet is printed; default `jsonl`.

Delivery checks#

Set with delivered.check. All with: values are templates, rendered per packet. Guide: Delivery checks.

Which output supports which check. pipo check rejects any other combination (P031).

Check	`sqlite`	`file`	`http`	`telegram`	`pipeline`	`stdout`
`ack`	✓	✓	✓	✓	✓	✓
`external`	✓	✓	✓	✓	✓	✓
`none`	✓	✓	✓	✓	✓	✓
`record_exists`	✓
`row_count`	✓
`query`	✓
`file_exists`		✓
`file_nonempty`		✓
`line_contains`		✓
`checksum`		✓
`status`			✓
`downstream`					✓
`follow_up`			✓

check: ack#

The connector's own success report is enough.

No with: settings.

check: external#

Wait for an outside acknowledgement.

No with: settings.

check: none#

Don't verify.

No with: settings.

check: record_exists#

A row matching where exists.

Field	Type	Required	Meaning
`where`	map	yes	Column → value pairs a row must match (with `=` and `AND`).

check: row_count#

query returns at least min rows.

Field	Type	Required	Meaning
`query`	string	yes	A read-only SELECT.
`params`	list		Values bound to the query's `?` placeholders.
`min`	integer or string		Rows the query must return at least: a number, or a template that renders to one; default 1. (≥ 0)

check: query#

Custom SQL returns a truthy first column.

Field	Type	Meaning
`sql`	string	A read-only SELECT whose first column in the first row must be truthy.
`query`	string	Alias of `sql`.
`params`	list	Values bound to the query's `?` placeholders.

One of sql and query is required.

check: file_exists#

The file exists.

Field	Type	Required	Meaning
`path`	string		File to look for; default the output's own `path`.

check: file_nonempty#

The file exists and is not empty.

Field	Type	Required	Meaning
`path`	string		File to look at; default the output's own `path`.

check: line_contains#

A line contains value.

Field	Type	Required	Meaning
`path`	string		File to search; default the output's own `path`.
`value`	string	yes	Text one line of the file must contain, e.g. `${meta.packet_id}`.

check: checksum#

The file hash equals sha256.

Field	Type	Required	Meaning
`path`	string		File to hash; default the output's own `path`.
`sha256`	string	yes	Expected SHA-256 of the file: 64 hex characters, any case.

check: status#

Write response status is in success.

Field	Type	Required	Meaning
`success`	list		Status codes that count as delivered, as `204`, `"2xx"` or `"200-299"`; default `2xx`.

check: downstream#

The receiving pipeline delivered the packet.

No with: settings.

check: follow_up#

A GET to url succeeds, or its body matches match.

Field	Type	Required	Meaning
`url`	string	yes	URL to GET after the write.
`headers`	map		Headers for the follow-up request.
`success`	list		Status codes that count as success; default `2xx`.
`match`	any		Text the body must contain, or an object that must be a subset of the JSON body.