Pipeline steps reference

Adapted from apps/gateway/docs/pipelines.md. Every example uses the real flat step shape. Validate any full config with cargo run -q -p mcpg --bin mcpg-config -- check <file>, and start from the backends reference for the pipeline backend itself.

A pipeline binding executes an ordered sequence of steps as a single MCP tool call. Steps share data through a pipeline context, and four step kinds can suspend execution to interact with the client. The gateway's PipelineStepConfig enum defines 24 step kinds.

Step shape

Every step is a flat object: a kind: discriminator, a unique id:, and that kind's own fields as siblings — not a nested per-kind sub-block. Backend steps flatten the same config as their standalone backend (an http step carries url: / method: directly).

backend:
  kind: pipeline
  pipeline_timeout_ms: 30000           # total timeout for all steps
  steps:
    - id: step1                        # unique within this pipeline
      kind: http                       # step kind discriminator
      url: "https://api.example.com/users"
      method: post
    - id: step2
      kind: transform
      expression: 'steps.step1.output.response.json.data'

Step IDs. Give each step an id of its own. A step's result is kept under its id, so a later step with the same id replaces it, and an expression that names a step that has not run fails the step that evaluates it.

Pipeline context

transform, cel_gate and input_transform are CEL expressions over these variables:

trust_level, principal_id and auth_provider are also bound at the top level, as policy expressions read them. ${env.NAME} and ${secret.NAME} in a step's fields are resolved when the config loads, not in these expressions.

Each step result:

{
  "step_id": {
    "output": { "...": "..." },
    "is_error": false,
    "duration_ms": 42
  }
}

The 24 step kinds

They group into backend steps (13), composition steps (3), notification steps (2), SQL container steps (2), and suspending steps (4).

Backend steps (13)

Backend steps execute the same adapters as standalone backends — the backend config flattens onto the step. A step's input is arguments, or what its optional input_transform CEL expression returns. In the step's own fields (url, headers, …), ${arguments.x} names a field of that input; those templates do not see steps, so an earlier step's output reaches a backend step through input_transform. See the backends reference for full field docs.

HTTP

- kind: http
  id: fetch_user
  url: "https://api.example.com/users"
  method: post
  timeout_ms: 5000
  expected_status_codes: [200]
  require_json_response: true
  input_transform: '{"id": arguments.user_id}'   # optional CEL expression: the JSON body

An http step's output is its call envelope: the body is steps.<id>.output.response.json (parsed) or steps.<id>.output.response.body (text).

Command

- kind: command
  id: run_script
  command: "/usr/bin/process"
  args: ["--json"]
  timeout_ms: 10000
  require_json_stdout: true

NATS

- kind: nats
  id: backend_query
  url: "nats://broker:4222"            # all nats steps/bindings must agree
  subject: "service.query"
  timeout_ms: 5000

Kafka

- kind: kafka
  id: kafka_request
  bootstrap_servers: "broker:9092"     # all kafka steps/bindings must agree
  group_id: "mcpg"
  request_topic: "requests"
  response_topic: "responses"
  timeout_ms: 10000

gRPC

- kind: grpc
  id: grpc_call
  url: "https://grpc.example.com"
  service: "example.UserService"
  method: "GetUser"
  timeout_ms: 5000

GraphQL

- kind: graphql
  id: graphql_query
  url: "https://api.example.com/graphql"
  operation: "query { users { id name } }"
  timeout_ms: 5000

SOAP

- kind: soap
  id: soap_call
  endpoint: "https://services.example.com/UserService"
  soap_version: "1.1"
  soap_action: "http://example.com/GetUser"
  body_template: |
    <GetUser xmlns="http://example.com/">
      <id>${arguments.user_id}</id>
    </GetUser>
  timeout_ms: 30000

LDAP

- kind: ldap
  id: lookup_user
  url: "ldaps://dc1.corp.example.com:636"
  bind_dn: "cn=svc-mcpg,ou=svc,dc=corp,dc=example,dc=com"
  bind_password: "${cred://vault-ad/svc-mcpg}"
  base_dn: "ou=people,dc=corp,dc=example,dc=com"
  scope: subtree
  filter: "(&(objectClass=person)(cn=*${arguments.q}*))"
  attributes: [cn, mail, department]
  timeout_ms: 10000

MSSQL

- kind: mssql
  id: lookup_employee
  host: "sql1.corp.example.com"
  database: "HR"
  user: "svc_mcpg"
  password: "${env.MSSQL_HR_PASSWORD}"
  op: query
  query: "SELECT id, full_name FROM dbo.employees WHERE id = @P1"
  params: ["arguments.id"]
  timeout_ms: 10000

AMQP

- kind: amqp
  id: enqueue_job
  uri: "amqps://mcpg:${env.RABBIT_PASSWORD}@rabbit.internal:5671/%2f"
  op: publish
  routing_key: "jobs.process"
  timeout_ms: 8000

Email

- kind: email
  id: notify
  op: send
  host: "smtp.corp.example.com"
  port: 587
  tls: starttls
  username: "svc-mcpg"
  password: "${env.SMTP_PASSWORD}"
  from: "alerts@corp.example.com"

SFTP

- kind: sftp
  id: fetch_drop
  op: get
  host: "sftp.partner.example.com"
  port: 22
  username: "svc-mcpg"
  password: "${env.SFTP_PASSWORD}"
  host_key_sha256: "SHA256:abc123…"
  root: "/outbound"

Mock

Flattens the mock backend config (response, delay_ms, error, error_message, passthrough).

- kind: mock
  id: fake_data
  response: { items: [1, 2, 3] }
  delay_ms: 100

Composition steps (3)

Transform

Produces a new value from a CEL expression over the pipeline context. The result becomes the step's output.

- kind: transform
  id: extract_name
  expression: 'steps.fetch_user.output.response.json.user.name'

Common patterns: extract nested fields (steps.step1.output.data.field), concatenate, or build map/list literals.

CEL gate (cel_gate)

Evaluates a CEL expression as a boolean guard. If false, the pipeline aborts with error_message (a default is used when omitted); if true, it continues.

- kind: cel_gate
  id: check_permission
  expression: 'steps.fetch_user.output.response.json.role == "admin"'
  error_message: "User does not have admin role"

Plugin transform (plugin_transform)

Reshapes the context by invoking a named transform plugin instead of an inline CEL expression — a generic bridge that works with any registered transform plugin. The first user is dev.mcpg.transform.jsonata, which evaluates a JSONata expression (stronger than CEL at building nested objects, projecting, and aggregating arrays).

- kind: plugin_transform
  id: reshape
  plugin: dev.mcpg.transform.jsonata
  config:
    expression: '{ "orderIds": steps.fetch.output.orders.id, "total": $sum(steps.fetch.output.orders.amount) }'

The plugin receives the full context (steps.<id>.output, arguments, context.*, tool_name) and its output becomes the step's output. The plugin must be loaded in plugins[]; a plugin error aborts the pipeline.

Notification steps (2)

Non-suspending steps that emit a server-to-client notification on the session's SSE channel and then continue immediately. Their data and message are sent as written: a ${…} in them other than ${env.NAME} and ${secret.NAME} stays as text.

Log

Emits a notifications/message.

- kind: log
  id: announce
  level: info
  data: "Charging the order"

Progress

Emits a notifications/progress. Silently skipped when the inbound request carried no progressToken.

- kind: progress
  id: halfway
  progress: 0.5
  message: "Validation complete"

SQL container steps (2)

Steps that operate against a named sql backend's connection pool. See the SQL backend for the backend itself.

SQL transaction (sql_tx)

Groups one or more SQL statements under a single transaction on a referenced SQL backend's pool — all-or-nothing. Reference the backend by name via backend:.

- kind: sql_tx
  id: charge_flow
  backend: orders_db                   # existing SQL backend name
  steps:
    - id: deduct_inventory
      sql: "UPDATE inventory SET qty = qty - 1 WHERE id = :id"
      params: [id]
    - id: record_order
      sql: "INSERT INTO orders (user_id, item_id) VALUES (:u, :i)"
      params: [u, i]
      row_mode: affected_rows

A failure at any nested step rolls the whole group back. Nested results surface downstream as steps.<tx_step_id>.output.steps.<nested_id>.

SQL await (sql_await)

Fire-and-wait against a referenced SQL backend whose profile declares an await block — runs the trigger, polls the check query, and evaluates the CEL predicate until it matches or times out. Same machinery as the standalone SQL await binding, exposed for pipeline composition.

- kind: sql_await
  id: wait_provisioned
  backend: provisioning_db             # existing SQL backend with an await block

Suspending steps (4)

Suspending steps interrupt execution to interact with the client. Pipeline state is persisted to the pipeline store, and a server-initiated JSON-RPC request is delivered to the client via the delivery bus. Execution resumes from the next step when the client responds; the client's response becomes the step's output.

Before suspending, the gateway checks the client's initialize capabilities. If the client does not support the requested interaction, the step returns an error instead of suspending. A message or a sampling message's content is sent as written, like the notification steps' text.

Elicitation

Prompts the client for input (elicitation/create).

- kind: elicitation
  id: ask_user
  message: "Please confirm the operation:"
  requested_schema:                    # optional JSON Schema for the response
    type: object
    properties:
      confirmed: { type: boolean }

Sampling

Requests the client to perform LLM sampling (sampling/createMessage).

- kind: sampling
  id: ai_analysis
  messages:
    - role: user
      content: "Summarize the open incidents in two sentences."
  max_tokens: 500

Roots list (roots_list)

Asks the client to enumerate its filesystem / URI roots (roots/list).

- kind: roots_list
  id: discover_roots
  timeout_ms: 30000

Gather

Emits several server-to-client input requests (any mix of elicitation, sampling, or roots) in one suspension and resumes once the client answers them together — distinct from listing the individual suspending steps in sequence, which suspends and resumes one at a time.

Note: inside gather.inputs[], the roots entry kind is roots (the standalone step uses roots_list).

- kind: gather
  id: collect_inputs
  inputs:                              # two or more entries, each with a correlation_token
    - kind: elicitation
      correlation_token: confirm
      message: "Confirm the operation?"
    - kind: sampling
      correlation_token: summary
      messages:
        - role: user
          content: "Summarize the change in one sentence."
      max_tokens: 200

Each answered input lands in this step's output under its correlation_token, so downstream transform steps read steps.collect_inputs.output.confirm, etc.

Execution lifecycle

For each step the executor evaluates input_transform (if any), runs the step adapter, stores the result in steps, and aborts if the step errored. A suspending step serializes state, sends the server request, and returns 202 Accepted; the client's response loads state and resumes from the next step. Exceeding pipeline_timeout_ms returns a timeout error.

What a pipeline returns

The result is the same whether the pipeline suspended on the way or not:

• All steps complete: the final step's output. A JSON object is the result's structuredContent and, serialized, its text; any other value is text only (a string as-is), because structuredContent is always an object.
• The final output carries __mcpg_verbatim_result: that CallToolResult as-is (content, structuredContent, isError), the same convention a backend binding honours. A backend that answers a domain error with HTTP 200 can mark it this way; for an http step, end the pipeline with a transform step that returns the response body (expression: steps.<id>.output.response.json).
• A backend step answers with an error result: that result, unchanged, with isError: true.
• Any other failure: isError: true, the text pipeline '<name>' failed at step '<id>': <reason>, and structuredContent {pipeline, failed_step, error, completed_steps, total_steps}.
• A cel_gate stops it: isError: true, the gate's message, and structuredContent {pipeline, gate_step, reason, completed_steps, total_steps}.

Distributed coordination

Suspending pipelines work across load-balanced gateway instances:

• Pipeline store persists execution state for suspension and resumption — InMemoryPipelineStore (single-instance default), NatsKvPipelineStore, or RedisPipelineStore. A state_version field implements compare-and-set fencing so only one instance can resume a given pipeline.
• Delivery bus routes server-initiated messages (elicitation / sampling requests, deferred results) to the instance holding the client's SSE stream (auto-selection: NATS > Redis > in-process). If no subscriber is connected, messages are stored as pending deliveries and replayed when the client reconnects.
• Pipeline reaper periodically cleans up pipeline states that exceed their pipeline_timeout_ms (default sweep: 30s).

Metrics

See also

• Backends reference — every backend plugin, including the pipeline backend and the SQL backend referenced by sql_tx / sql_await.
• Configuration reference — the full AppConfig key tree.