Agents as operators

Pipo is agent-first: an agent can write a pipeline, run it, watch it, step in when something goes wrong, change it while it runs, and step out again. It uses the same documented interfaces a human uses, with no workarounds. This page covers how agents operate pipelines through the MCP endpoint, and what each pipeline lets them do through its agent: policy.

Four roles#

Agents never change Pipo's own code at runtime. They change .pipo files, through a controlled protocol.

Quick start: connect Claude Code#

1. Add the engine to Claude Code. The engine listens on port 4747 when the CLI starts it (pipo engine start, pipo ui, or any command):

   Copy

   pipo engine start
   claude mcp add --transport http pipo http://127.0.0.1:4747/mcp
   

2. Approve it in the browser. In Claude Code, run /mcp and pick pipo. Your browser opens the Pipo dashboard on a 🔑 Connect an agent page (log in as the owner first if you aren't): pick what it may do (its scope), which pipelines it may see, and the name its actions are recorded under, then Approve. Claude Code gets a token and refreshes it by itself.

3. Let the pipeline be operated by agents, in its .pipo file:

   Copy

   agent:
     control: true
     actions: [pause, resume, replay, push]
   

The dashboard's 🔑 Agents page lists the connected agents, and revokes them. pipo mcp grants does the same from a terminal. The Pipo plugin sets up the server and adds a Pipo skill in one step.

Any MCP client that speaks Streamable HTTP and the MCP authorization flow works the same way.

Static tokens#

For a client without OAuth, or a script, list a token in <home>/config.yaml (~/.pipo/config.yaml by default) and restart the engine:

engine:
  mcp:
    tokens:
      - name: ops-agent                # the name its actions are recorded under
        token: env:PIPO_MCP_TOKEN      # a reference, never the token itself
        scope: operate                 # read | operate | edit | author
        pipelines: [people-intake]     # optional allowlist

export PIPO_MCP_TOKEN=$(openssl rand -hex 24)   # at least 16 characters
pipo engine stop; pipo engine start
claude mcp add --transport http pipo http://127.0.0.1:4747/mcp \
  --header "Authorization: Bearer $PIPO_MCP_TOKEN"

The MCP endpoint#

The engine serves an MCP server at /mcp on its gateway (port 4747 by default, or engine.listen, on 127.0.0.1 unless engine.host exposes it).

• Transport. Streamable HTTP without sessions: each POST carries one JSON-RPC message and gets a JSON answer. Notifications get 202. There is no server stream, so GET and DELETE are 405. Batches are refused. Revisions 2025-06-18 and 2025-11-25 are supported.
• Methods. initialize, ping, tools/list, tools/call, resources/list, resources/templates/list, resources/read.
• Addressed to this engine. Like /api, it answers only requests addressed to a loopback host (127.0.0.1, localhost, [::1]) or to the host of engine.public_url (see Owner and remote access). Any other Origin is 403, and bodies over 4 MiB are 413.
• The same operations are available over REST and through pipo … --json. MCP tools call the REST handlers, so their answers, errors and hints are the same.

Tokens and scopes#

Every request sends Authorization: Bearer <token>: an OAuth access token, or a static token from config.yaml. A missing or unknown token is 401, with WWW-Authenticate: Bearer resource_metadata="…" so the client can start the OAuth flow, and a hint.

OAuth#

The engine is its own OAuth server, following the MCP authorization spec:

• Discovery. /.well-known/oauth-protected-resource names the engine as the authorization server, and /.well-known/oauth-authorization-server lists its endpoints.
• Registration. Clients register themselves at /oauth/register. Only public clients, with redirect URIs on localhost, 127.0.0.1 or [::1] (any port), or https://.
• Approval. /oauth/authorize (with PKCE) sends your browser to the dashboard's consent page, which needs the owner's login, and the grant records who approved it. It shows the client's name, which the client chose itself, and where the token will go. You pick the scope, all pipelines or some, and the name. An author scope is flagged: it lets the agent write pipelines and start them once approved.
• Tokens. Access tokens last an hour and refresh tokens 30 days, rotated on every use. Only their SHA-256 digests are kept, in <home>/mcp-grants.json, so grants survive engine restarts.
• Revoking. The Agents page, pipo mcp grants revoke <id>, or POST /api/mcp/grants/<id>/revoke. The agent's next call is 401, and its client asks you again.

Reconnecting under the same name keeps the pipelines the agent wrote (see Agents that build pipelines).

Static tokens and scopes#

engine.mcp.tokens is a list:

Scopes are the same for OAuth grants and static tokens:

What a pipeline allows: the policy#

A token's scope is only half of it. The pipeline's agent: block (in its version in force) applies on top, checked in this order: the scope, the allowlist, then the policy.

• Every pipeline tool and resource needs agent.control: true. Without it, the pipeline is listed by list_pipelines with agent_control: false, and nothing else works on it.
• pause, resume, replay, rerun, push and ack also need that action in agent.actions. So do start_pipeline and stop_pipeline (start, stop) on a pipeline the agent didn't write itself.
• propose_change and rollback are checked against agent.edit (see Live changes).
• resolve isn't gated by agent.actions: then: agent and agent.on_stall: handle are the grant, for the packets they hand over.

Tools#

Every pipeline tool takes pipeline (the name). Results are JSON text. A refused call is a tool result with isError: true and {error, hint, code}.

The authoring tools, create_pipeline, test_pipeline, start_pipeline, stop_pipeline and list_approvals, are described in Agents that build pipelines.

Unknown arguments are refused, and null counts as absent. An agent's replay, rerun, resolve, proposals and rollbacks are recorded with the token's name, and author_kind/by_kind agent.

Resources#

Pipeline resources follow the same allowlist and agent.control rules as the tools. resources/list lists only the pipelines the token can use.

Redaction#

Every answer is redacted before it leaves the engine:

• secrets, always: the pipeline's secrets, and the MCP token values;
• paths in packet data listed in agent.redact. Their values are replaced with [redacted] in every payload, and then masked wherever else they appear in the answer, such as in log lines, errors or event details.

agent.redact paths are dotted, * matches one segment, and a leading data. is optional. A path that meets a list applies to each element.

agent:
  control: true
  redact: [data.email, data.contacts.*.phone]

The agent: policy block#

agent:
  control: true                        # enable the agent endpoint for this pipeline
  actions: [pause, resume, replay, push]
  edit:                                # paths an agent may change; everything else is forbidden
    - nodes.*.with.prompt
    - nodes.normalize
    - errors
  redact: [data.email]
  on_stall: notify                     # notify | handle
  verify: last 20                      # dry-run a proposal on the last 20 delivered packets

The policy that applies is always the one in the version in force, never the one in a proposal. So an agent can't widen its own permissions: changes under agent are forbidden to agents anyway.

Handing packets to the agent#

A pipeline can hand problem packets to an agent instead of dead-lettering them. The packet is escalated: it stops where it was, keeps its data and version, and waits, across restarts, until it is resolved. Nothing is dropped while it waits.

nodes:
  enrich:
    from: input
    transform: http
    with: { url: "https://api.example.com/people/${data.id}" }
    on_error:
      retry: 3
      then: agent          # after the retries, hand it to the agent
delivered:
  stall:
    after: 10m
    then: agent            # flag the stall, and tell the agent
agent:
  control: true
  on_stall: handle         # the agent takes over the stalled packets

Escalated packets count as pending, including toward buffer.max. So if no agent ever comes, the input fills up and then pushes back (HTTP answers 503). They don't make a stall themselves, and a drain doesn't wait for them.

Resolving#

An agent (or a human) settles escalated packets with one of three actions:

pipo packets people-intake --state escalated          # find them
pipo resolve people-intake 01J9Z… --action retry --reason "API is back"

The MCP tool is resolve, and REST is POST /api/pipelines/<name>/resolve. A list of ids is all or nothing: every id is checked first, then all of them commit together. A packet that fanned out is resolved per copy, and the error's hint lists the copies that are waiting.

A typical agent loop: get_status shows escalated > 0, inspect_packet with state: escalated lists them, inspect_packet with a packet_id shows the error, the agent fixes the cause (a proposal, or waiting for an outside system), then resolve with retry.

Audit#

Everything an agent does is recorded in the pipeline's journal with its token's name: pauses (reason agent), replays, resolves (by, by_kind: agent, reason) and versions (author, author_kind: agent, reason, proposal). pipo history and the dashboard's agent feed show them.

See also#

• Live changes and proposals
• REST, SSE and MCP
• Configuration
• The spec: §9 Agents in Pipo