Overview

MPE Flow is a multi-protocol workflow orchestration engine. It runs as either a desktop GUI or a CLI tool — from the same binary.

Two Modes, One Binary

No arguments → Opens the Tauri desktop GUI
With arguments → Runs in CLI mode

On Linux, only the CLI binary is available — no GUI dependencies required.

Installation

Windows & macOS

Download the installer from the Changelog page. The same binary supports both GUI and CLI modes.

Linux (CLI)

Download the pre-built binary and add it to your PATH:

# Download and extract
curl -LO https://download.mpe.run/releases/latest/mpe-vX.Y.Z-linux-x86_64.tar.gz
tar xzf mpe-vX.Y.Z-linux-x86_64.tar.gz

# Make executable and move to PATH
chmod +x mpe
sudo mv mpe /usr/local/bin/

# Verify
mpe --help

Note

The Linux build is CLI-only — no GUI, no GTK/WebKit dependencies required.

Quick Example

Create a file hello.mpf:

{
  "_version": 1,
  "flow": {
    "uuid": "example-flow",
    "name": "Hello World",
    "nodes": [
      {
        "uuid": "entry-1",
        "type": "entry",
        "name": "Start",
        "data": { "type": "entry" }
      },
      {
        "uuid": "http-1",
        "type": "http",
        "name": "Fetch Data",
        "data": {
          "type": "http",
          "url": "https://httpbin.org/get",
          "method": "get"
        }
      },
      {
        "uuid": "end-1",
        "type": "end",
        "name": "End",
        "data": { "type": "end" }
      }
    ],
    "connections": [
      {
        "id": "conn-1",
        "source_node_uuid": "entry-1",
        "target_node_uuid": "http-1",
        "source_port_id": "out",
        "target_port_id": "in"
      },
      {
        "id": "conn-2",
        "source_node_uuid": "http-1",
        "target_node_uuid": "end-1",
        "source_port_id": "true",
        "target_port_id": "in"
      }
    ]
  }
}

Run it:

mpe run -f hello.mpf

CLI 命令参考

命令	说明	示例
`mpe run`	执行工作流文件 (.mpf)	`mpe run -f flow.mpf`
`mpe validate`	校验工作流结构（不执行）	`mpe validate -f flow.mpf`
`mpe debug`	单步调试输出（NDJSON 协议）	`mpe debug -f flow.mpf`
`mpe stress`	运行压力测试	`mpe stress run -f flow.mpf`
`mpe flow`	工作流管理（list, show, create, delete）	`mpe flow list`
`mpe report`	报告管理（list, show, delete）	`mpe report list`
`mpe run-node`	单节点协议连通性验证	`mpe run-node '{"type":"redis:connect","host":"127.0.0.1","port":6379}'`
`mpe plugin`	插件管理（dir, list, install）	`mpe plugin list`

# 查看所有命令
mpe --help

# 查看特定命令帮助
mpe run --help

CI/CD 集成 (GitHub Actions)

mpe run 支持将执行结果导出为标准 JUnit XML 格式，可直接被 Jenkins、GitHub Actions 和 GitLab CI 等持续集成平台识别与解析。工作流中的每个节点对应一个 <testcase>；执行失败和跳过的节点分别映射为 <failure> 和 <skipped> 元素。

# 导出 JUnit XML 报告供 CI 使用
mpe run -f flow.mpf --report-format junit --report-file results.xml

# 或者直接将 JUnit XML 输出到 stdout（不带 --report-file）
mpe run -f flow.mpf --report-format junit

在仓库中添加工作流步骤，通过 dorny/test-reporter 在 Pull Request 中自动执行工作流并展示测试报告：

name: Flow Tests

on:
  pull_request:

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Run MPE flows
        run: |
          mpe run -f tests/flows/api-smoke.mpf --report-format junit --report-file results.xml

      - name: Publish test report
        if: always()
        uses: dorny/test-reporter@v1
        with:
          name: MPE Flow Results
          path: results.xml
          reporter: java-junit
          fail-on-error: 'true'

提示

JUnit 报告如实反映节点级别的测试结果：断言、HTTP、WebSocket 等协议节点映射为独立的测试用例。任何节点执行失败都会导致非零退出码，因此 mpe run --report-format junit --report-file results.xml 默认会在失败时使 CI Job 报错中断。

Output Format

All commands output JSON to stdout:

{
  "success": true,
  "data": null,
  "error": null,
  "execution_time": 156,
  "node_reports": [
    {
      "node_uuid": "entry-001",
      "node_type": "entry",
      "node_name": "Start",
      "status": "success",
      "duration_ms": 1,
      "used_port_id": "out"
    },
    {
      "node_uuid": "http-001",
      "node_type": "http",
      "node_name": "Fetch Data",
      "status": "success",
      "duration_ms": 233,
      "output_data": {
        "success": true,
        "status": 200,
        "body": { ... },
        "headers": { ... }
      }
    }
  ],
  "flow_report": {
    "flow_name": "Hello World",
    "total_nodes": 3,
    "executed_count": 3,
    "status": "success",
    "duration_ms": 234
  }
}

Exit Codes

Code	Meaning
`0`	Success
`1`	Failure (error details in stderr and JSON output)
`2`	Invalid arguments

Flow File Structure

MPE flow files use the .mpf extension and follow the FlowFile wrapper format:

{
  "_version": 1,
  "flow": {
    "uuid": "my-flow",
    "name": "My Flow",
    "initial_variables": {
      "api_base": "https://api.example.com",
      "token": "abc123"
    },
    "nodes": [ ... ],
    "connections": [ ... ]
  }
}

Top-level Fields

Field	Type	Required	Description
`_version`	number	No	File format version (default: 1)
`flow`	object	Yes	Flow data container

Flow Fields

Field	Type	Required	Description
`uuid`	string	Yes	Unique flow identifier
`name`	string	Yes	Human-readable flow name
`initial_variables`	object	No	Variables available via `{{key}}` syntax
`nodes`	array	Yes	List of nodes
`connections`	array	Yes	Connections between nodes

Node Structure

{
  "uuid": "node-uuid",
  "type": "http",
  "name": "My HTTP Request",
  "data": {
    "type": "http",
    "url": "https://api.example.com",
    "method": "get"
  }
}

Field	Type	Required	Description
`uuid`	string	Yes	Unique node identifier
`type`	string	Yes	Node type (e.g. `http`, `entry`)
`name`	string	Yes	Display name
`data`	object	Yes	Node configuration. Must include `"type"` matching the node type
`on_error`	string	No	Error strategy: `"route_to_false"` (default), `"ignore"`, `"abort_flow"`

Important: data.type Field

Every data object must include "type": "<node_type>" for Rust deserialization. The value must match the node's top-level type field.

Connections

Connections define the execution flow between nodes.

{
  "id": "conn-1",
  "source_node_uuid": "entry-1",
  "target_node_uuid": "http-1",
  "source_port_id": "out",
  "target_port_id": "in"
}

Field	Description
`source_node_uuid`	UUID of the source node
`target_node_uuid`	UUID of the target node
`source_port_id`	Output port: `out` (single output), `true`/`false` (dual output)
`target_port_id`	Must be `"in"` for all nodes

Critical Port Rules

target_port_id must always be "in" (not "input")
Nodes with dual output (true/false) should have both ports connected to prevent flow termination on failure
entry uses out, end has no output

Port Reference

Node Type	Input	Output
`entry`	none	`out`
`end`	`in`	none
`http`	`in`	`true` / `false`
`conditional`	`in`	`true` / `false`
`script`	`in`	`true` / `false`
`assertion`	`in`	`true` / `false`
`variable_extractor`	`in`	`out`
`ws_`, `tcp_`, `udp_*`, `sse_connect`/`sse_listen`, `graphql_connect`/`query`/`subscribe`/`introspect`	`in`	`true` / `false`
`sse_disconnect`, `graphql_disconnect`, `proxy`	`in`	`out`

Variables

Variable Pool vs Node Output

{{var}} syntax can only access the Variable Pool. Protocol node outputs are stored separately in last_node_output and do NOT automatically become variables.

To use node output data with {{var}}, you must first extract it using a variable_extractor or script node.

Syntax

Syntax	Description
`{{variable_name}}`	Reference a variable from the pool
`{{obj.field}}`	Access nested field with dot notation

Adding Variables to the Pool

Initial variables: Define in flow.initial_variables
variable_extractor: Extract from last_node_output using JSONPath
script: Use fn.variables.set(name, value)

// variable_extractor example
{
  "type": "variable_extractor",
  "data": {
    "type": "variable_extractor",
    "output_mappings": [
      { "source": "$.body.id", "target": "user_id" },
      { "source": "$.status", "target": "http_status" }
    ]
  }
}

// Later use: {{user_id}}, {{http_status}}

Conditional / Assertion Paths

conditional and assertion nodes read directly from last_node_output:

Node	Path Format	Example
`conditional.field_path`	Plain name, `{{var}}`, or `$.path`	`status`, `$.body.id`
`assertion.target`	JSONPath with `$.` prefix (required)	`$.status`, `$.body.data.id`

Limitations

Only dot notation: {{obj.field.subfield}}
No array indexing: {{items[0]}} — not supported
No function calls: {{func()}} — not supported

Entry & End Nodes

Entry Node

Starting point of every flow. Exactly one required.

{
  "uuid": "entry-1",
  "type": "entry",
  "name": "Start",
  "data": { "type": "entry" }
}

End Node

Terminal node. Flow ends when reaching end.

{
  "uuid": "end-1",
  "type": "end",
  "name": "End",
  "data": { "type": "end" }
}

HTTP Node

Send HTTP requests. Supports GET, POST, PUT, DELETE, PATCH, HEAD, OPTIONS.

{
  "uuid": "http-1",
  "type": "http",
  "name": "API Call",
  "data": {
    "type": "http",
    "url": "{{api_base}}/users",
    "method": "post",
    "headers": {
      "Authorization": "Bearer {{token}}"
    },
    "body_type": "body",
    "content_type": "json",
    "body": "{\"name\": \"test\"}",
    "timeout_ms": 30000
  }
}

Core Fields

Field	Type	Default	Description
`url`	string	—	Request URL (supports `{{var}}`)
`method`	string	`get`	HTTP method
`headers`	object	`{}`	Request headers
`body_type`	string	`none`	`none` / `body` / `form-data` / `x-www-form-urlencoded` / `binary`
`content_type`	string	`json`	`json` / `text` / `html` / `xml`
`timeout_ms`	number	`30000`	Timeout (1000–300000ms)

HTTP Output

{
  "success": true,
  "status": 200,
  "body": { "id": 1, "name": "test" },
  "headers": { "content-type": "application/json" },
  "timing": { "connect_ms": 45, "total_ms": 150 }
}

Conditional Node

Branch execution based on conditions. Routes to true or false port.

Simple Condition

{
  "uuid": "cond-1",
  "type": "conditional",
  "name": "Check Status",
  "data": {
    "type": "conditional",
    "condition_type": "simple",
    "field_path": "status",
    "operator": "equal",
    "expected_value": 200
  }
}

Advanced Expression

{
  "data": {
    "type": "conditional",
    "condition_type": "advanced",
    "advanced_expression": "status >= 200 && status < 300"
  }
}

Operators

Operator	Description
`equal`	Equals
`not_equal`	Not equals
`greater`	Greater than
`greater_or_equal`	Greater or equal
`less`	Less than
`less_or_equal`	Less or equal
`contains`	Contains substring
`starts_with`	Starts with
`ends_with`	Ends with
`exists`	Field exists

Script Node

Execute JavaScript code in a QuickJS runtime (ES2020 subset).

{
  "uuid": "script-1",
  "type": "script",
  "name": "Process Data",
  "data": {
    "type": "script",
    "script": "const data = JSON.parse(fn.response.raw); fn.variables.set('user_id', data.body.id);",
    "timeout_ms": 5000
  }
}

Script API

API	Description
`fn.variables.get(name)`	Read variable from pool
`fn.variables.set(name, value)`	Write variable to pool
`fn.response.raw`	Full upstream node output (JSON string)
`fn.response.code`	HTTP status code (HTTP nodes only)
`fn.console.log(msg)`	Log to execution output
`fn.flow.stop(reason)`	Stop flow execution
`fn.util.encodeBase64(str)`	Base64 encode
`fn.util.md5(str)`	MD5 hash
`fn.util.sha256(str)`	SHA-256 hash
`fn.util.uuid()`	Generate UUID v4
`fn.util.sleep(ms)`	Pause execution

Assertion Node

Validate conditions with detailed pass/fail reporting.

{
  "uuid": "assert-1",
  "type": "assertion",
  "name": "Validate Response",
  "data": {
    "type": "assertion",
    "mode": "all",
    "assertions": [
      { "id": "a1", "description": "Status is 200", "target": "$.status", "operator": "eq", "expected": 200 },
      { "id": "a2", "description": "Has data", "target": "$.body.data", "operator": "exists" }
    ]
  }
}

Assertion Operators

Category	Operators
Comparison	`eq`, `ne`
Numeric	`gt`, `gte`, `lt`, `lte`
String	`contains`, `not_contains`, `matches`
Existence	`exists`, `not_exists`
Collection	`in`, `not_in`
Type	`type_is`
Range	`between`

Target Path Format

Assertion target must use JSONPath with $. prefix (e.g. $.status, $.body.id). Do not add body. prefix for non-HTTP nodes.

WebSocket Nodes

Three nodes for WebSocket communication: connect, send/collect, close.

// Connect
{ "type": "ws_connect", "data": { "type": "ws_connect", "url": "wss://echo.example.com/ws" } }

// Send and collect response
{ "type": "ws_send_collect", "data": {
    "type": "ws_send_collect",
    "connection_id": "ws-connect-uuid",
    "message": "{\"type\":\"ping\"}",
    "collect_timeout_ms": 10000
  }
}

// Close
{ "type": "ws_close", "data": { "type": "ws_close", "connection_id": "ws-connect-uuid" } }

Other Protocols

TCP

{ "type": "tcp_connect", "data": { "type": "tcp_connect", "host": "localhost", "port": 8080 } }
{ "type": "tcp_send", "data": { "type": "tcp_send", "connection_id": "uuid", "data": "Hello" } }
{ "type": "tcp_receive", "data": { "type": "tcp_receive", "connection_id": "uuid" } }
{ "type": "tcp_close", "data": { "type": "tcp_close", "connection_id": "uuid" } }

UDP

{ "type": "udp_bind", "data": { "type": "udp_bind", "local_addr": "0.0.0.0", "local_port": 8888 } }
{ "type": "udp_send_to", "data": { "type": "udp_send_to", "connection_id": "uuid", "data": "Hello", "target_addr": "192.168.1.100", "target_port": 9999 } }
{ "type": "udp_recv_from", "data": { "type": "udp_recv_from", "connection_id": "uuid" } }
{ "type": "udp_close", "data": { "type": "udp_close", "connection_id": "uuid" } }

SSE (Server-Sent Events)

{ "type": "sse_connect", "data": { "type": "sse_connect", "url": "https://api.example.com/events" } }
{ "type": "sse_listen", "data": { "type": "sse_listen", "connection_id": "uuid", "max_events": 100 } }
{ "type": "sse_disconnect", "data": { "type": "sse_disconnect", "connection_id": "uuid" } }

GraphQL

{ "type": "graphql_connect", "data": { "type": "graphql_connect", "endpoint": "https://api.example.com/graphql" } }
{ "type": "graphql_query", "data": { "type": "graphql_query", "connection_id": "uuid", "query": "query { users { id name } }" } }
{ "type": "graphql_subscribe", "data": { "type": "graphql_subscribe", "connection_id": "uuid", "query": "subscription { newUser { id } }" } }
{ "type": "graphql_introspect", "data": { "type": "graphql_introspect", "connection_id": "uuid" } }
{ "type": "graphql_disconnect", "data": { "type": "graphql_disconnect", "connection_id": "uuid" } }

Plugin Protocols

More protocols (Redis, MySQL/PostgreSQL, MongoDB, SMTP, gRPC, MCP) are available as installable plugins — see Plugin Development below.

Plugin Overview

Plugins extend MPE Flow with new protocol node types. A plugin runs as a sidecar process that communicates with the host over stdio using JSON-RPC 2.0 — one JSON message per line (LF-delimited, CRLF tolerated). At startup the host spawns each plugin process and performs a describe handshake, registering the declared node types into the shared NodeRegistry. Plugin nodes behave identically to built-in nodes in flows, the executor, and execution reports.

Installation

Place a plugin directory containing a plugin.json manifest into the plugin directory (default get_data_dir()/plugins; %APPDATA%/multi-protocol-flow-executor/plugins on Windows; override with the MPE_PLUGIN_DIR environment variable). The host scans synchronously at startup: invalid manifests are skipped without crashing, and a plugin whose describe fails 3 consecutive times is quarantined for the host's lifetime.

plugin.json Fields

Field	Description
`name`	Plugin name (convention: directory name)
`version`	Semantic version
`description`	Human-readable description
`min_host_version`	Optional version gate for host compatibility
`entry`	`command` + `args` used to launch the plugin; interpreter plugins are first-class citizens (e.g. `"python"` + `["plugin.py"]`)
`env`	Environment variables injected into the plugin process
`permissions`	Reserved; ignored in P0
`capabilities.streaming`	`true` = resident process; `false` = spawned on demand and recycled after 60s idle
`capabilities.single_node`	Node-level capability; `true` = runnable standalone via `mpe run-node` for connectivity verification
`locales` / `default_locale`	UI language declarations; the host injects the active locale via `MPE_LOCALE`

Protocol Plugins in This Repository

The plugins/ directory ships protocol plugins for: amqp, db (SQLite/MySQL/PostgreSQL), grpc, imap, kafka, mcp, mongo, mqtt, redis, smtp. Install by building a plugin and placing its directory into the plugin directory, or install from the Plugins page.

SDK & Development

The SDK and wire contract live in the standalone public repository multi-protocol-flow/mpe-plugin-sdk. The Rust SDK is first-class; lightweight Python and Node templates are also provided. Wire types are shared from mpe-plugin-sdk::protocol, so host and plugin never drift.

Development Flow

Clone the SDK repository
Implement the Plugin trait (describe + execute; optional flow_ended to release per-execution resources)
Build the sidecar binary
Place it into the plugin directory

Reference Docs

SDK repository: docs/plugin-guide.md (EN) / docs/plugin-guide.zh-CN.md (中文) and docs/plugin-protocol.md
Host repository: docs/plugin-architecture.md, docs/plugin-perf.md (visible under the multi-protocol-flow GitHub organization)

Marketplace

Plugins can be installed from the GUI settings panel's Plugin Market tab, or from the CLI:

mpe plugin dir
mpe plugin list
mpe plugin install <name> [--version <v>] [--registry <url>] [--dir <dir>] [--force]

Registry Contract

GET {base}/plugins returns the plugin list; each entry has a platform asset (zip) url and an optional sha256 — when provided the host enforces checksum validation and refuses mismatched installs.
Platform keys: windows-x64 | windows-arm64 | linux-x64 | linux-arm64 | macos-x64 | macos-arm64.
The zip root must contain a single top-level directory named after the plugin, holding plugin.json and the plugin files.

The registry base URL is configured via the MPE_PLUGIN_REGISTRY environment variable or the --registry flag. The public registry is served from GitHub Pages (multi-protocol-flow.github.io/mpe-plugin-registry) as a static JSON implementing this contract — browse and install plugins on the Plugins page.