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://mpe-downloads.sunanzhi.com/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 Commands

Command Description Example
mpe run Execute a flow mpe run -f flow.mpf
mpe validate Validate flow structure without executing mpe validate -f flow.mpf
mpe debug Run with debug output (NDJSON protocol) mpe debug -f flow.mpf
mpe stress Run stress test mpe stress run -f flow.mpf
mpe flow Manage flows (list, show, create, delete) mpe flow list
mpe report Manage reports (list, show, delete) mpe report list 
# View all commands
mpe --help

# View command-specific help
mpe run --help

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)

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
entrynoneout
endinnone
httpintrue / false
conditionalintrue / false
scriptintrue / false
assertionintrue / false
variable_extractorinout
ws_*, db_*, grpc_*, etc.intrue / false
db_close, ws_close, redis_close, etc.inout

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

  1. Initial variables: Define in flow.initial_variables
  2. variable_extractor: Extract from last_node_output using JSONPath
  3. 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
equalEquals
not_equalNot equals
greaterGreater than
greater_or_equalGreater or equal
lessLess than
less_or_equalLess or equal
containsContains substring
starts_withStarts with
ends_withEnds with
existsField 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.rawFull upstream node output (JSON string)
fn.response.codeHTTP 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
Comparisoneq, ne
Numericgt, gte, lt, lte
Stringcontains, not_contains, matches
Existenceexists, not_exists
Collectionin, not_in
Typetype_is
Rangebetween

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" } }

Database Nodes

Support PostgreSQL, MySQL, SQLite. Use connection_uuid (not connection_id).

// Connect
{ "type": "db_connect", "data": {
    "type": "db_connect",
    "connection_uuid": "pg-conn",
    "driver": "postgres",
    "host": "localhost",
    "port": 5432,
    "database": "mydb",
    "username": "user",
    "password": "pass"
  }
}

// Query
{ "type": "db_query", "data": {
    "type": "db_query",
    "connection_uuid": "pg-conn",
    "sql": "SELECT * FROM users WHERE id = $1",
    "params": ["123"]
  }
}

// Transaction
{ "type": "db_begin", "data": { "type": "db_begin", "connection_uuid": "pg-conn" } }
{ "type": "db_commit", "data": { "type": "db_commit", "connection_uuid": "pg-conn" } }
{ "type": "db_close", "data": { "type": "db_close", "connection_uuid": "pg-conn" } }

gRPC Nodes

// Connect
{ "type": "grpc_connect", "data": { "type": "grpc_connect", "url": "https://localhost:50051", "use_tls": false } }

// Call
{ "type": "grpc_call", "data": {
    "type": "grpc_call",
    "connection_id": "grpc-connect-uuid",
    "service_name": "helloworld.Greeter",
    "method_name": "SayHello",
    "request_json": "{\"name\":\"world\"}"
  }
}

// Close
{ "type": "grpc_close", "data": { "type": "grpc_close", "connection_id": "grpc-connect-uuid" } }

MQTT Nodes

Supports MQTT 3.1.1 / 5.0, TLS, WebSocket.

// Connect
{ "type": "mqtt_connect", "data": { "type": "mqtt_connect", "url": "mqtt://localhost:1883", "client_id": "mpe-client" } }

// Publish
{ "type": "mqtt_publish", "data": {
    "type": "mqtt_publish",
    "connection_id": "mqtt-connect-uuid",
    "topic": "sensors/temp",
    "payload": "{\"value\":25.5}",
    "qos": 1
  }
}

// Subscribe
{ "type": "mqtt_subscribe", "data": {
    "type": "mqtt_subscribe",
    "connection_id": "mqtt-connect-uuid",
    "topic": "sensors/#",
    "max_messages": 10
  }
}

// Disconnect
{ "type": "mqtt_disconnect", "data": { "type": "mqtt_disconnect", "connection_id": "mqtt-connect-uuid" } }

Kafka Nodes

// Connect
{ "type": "kafka_connect", "data": { "type": "kafka_connect", "bootstrap_servers": "localhost:9092" } }

// Produce
{ "type": "kafka_produce", "data": {
    "type": "kafka_produce",
    "connection_id": "kafka-connect-uuid",
    "topic": "my-topic",
    "value": "{\"name\":\"Alice\"}"
  }
}

// Consume
{ "type": "kafka_consume", "data": {
    "type": "kafka_consume",
    "connection_id": "kafka-connect-uuid",
    "topic": "my-topic",
    "group_id": "my-group",
    "max_messages": 1
  }
}

// Close
{ "type": "kafka_close", "data": { "type": "kafka_close", "connection_id": "kafka-connect-uuid" } }

Redis Nodes

// Connect
{ "type": "redis_connect", "data": { "type": "redis_connect", "host": "127.0.0.1", "port": 6379 } }

// Command
{ "type": "redis_command", "data": {
    "type": "redis_command",
    "connection_id": "redis-connect-uuid",
    "command": "SET",
    "args": ["mykey", "Hello Redis"]
  }
}

// Close
{ "type": "redis_close", "data": { "type": "redis_close", "connection_id": "redis-connect-uuid" } }

SMTP & IMAP Nodes

SMTP (Send Email)

// Connect
{ "type": "smtp_connect", "data": {
    "type": "smtp_connect",
    "host": "smtp.gmail.com",
    "port": 587,
    "tls_mode": "starttls",
    "auth": { "type": "plain", "username": "user@example.com", "encrypted_password": "xxx" }
  }
}

// Send
{ "type": "smtp_send", "data": {
    "type": "smtp_send",
    "connection_id": "smtp-connect-uuid",
    "from": "sender@example.com",
    "to": ["recipient@example.com"],
    "subject": "Test Email",
    "body": "Hello World"
  }
}

// Disconnect
{ "type": "smtp_disconnect", "data": { "type": "smtp_disconnect", "connection_id": "smtp-connect-uuid" } }

IMAP (Read Email)

// Connect
{ "type": "imap_connect", "data": {
    "type": "imap_connect",
    "host": "imap.gmail.com",
    "port": 993,
    "tls_mode": "implicit",
    "auth": { "type": "plain", "username": "user@example.com", "encrypted_password": "xxx" }
  }
}

// Select mailbox
{ "type": "imap_select", "data": { "type": "imap_select", "connection_id": "imap-connect-uuid", "mailbox": "INBOX" } }

// Search
{ "type": "imap_search", "data": { "type": "imap_search", "connection_id": "imap-connect-uuid", "criteria": "UNSEEN" } }

// Fetch
{ "type": "imap_fetch", "data": { "type": "imap_fetch", "connection_id": "imap-connect-uuid", "uids": "1:5" } }

// Close
{ "type": "imap_close", "data": { "type": "imap_close", "connection_id": "imap-connect-uuid" } }

Auth Format

SMTP and IMAP auth must be an object with "type" field:

{"type": "plain", "username": "xxx", "encrypted_password": "xxx"}

Not "password" — the field name is encrypted_password.

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_disconnect", "data": { "type": "graphql_disconnect", "connection_id": "uuid" } }

MCP (Model Context Protocol)

// HTTP transport
{ "type": "mcp_connect", "data": { "type": "mcp_connect", "url": "https://mcp.example.com", "transport": "streamable_http" } }

// Stdio transport
{ "type": "mcp_connect", "data": { "type": "mcp_connect", "transport": "stdio", "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem"] } }

{ "type": "mcp_call_tool", "data": { "type": "mcp_call_tool", "connection_id": "uuid", "tool_name": "get_weather", "arguments": {"city": "Beijing"} } }
{ "type": "mcp_close", "data": { "type": "mcp_close", "connection_id": "uuid" } }

AMQP (RabbitMQ)

{ "type": "amqp_connect", "data": { "type": "amqp_connect", "url": "amqp://guest:guest@localhost:5672/%2f" } }
{ "type": "amqp_exchange_declare", "data": { "type": "amqp_exchange_declare", "connection_id": "uuid", "exchange": "my.exchange", "exchange_type": "direct" } }
{ "type": "amqp_queue_declare", "data": { "type": "amqp_queue_declare", "connection_id": "uuid", "queue": "my.queue" } }
{ "type": "amqp_publish", "data": { "type": "amqp_publish", "connection_id": "uuid", "exchange": "my.exchange", "routing_key": "my.rk", "body": "{\"msg\":\"hello\"}" } }
{ "type": "amqp_consume", "data": { "type": "amqp_consume", "connection_id": "uuid", "queue": "my.queue", "max_messages": 10 } }
{ "type": "amqp_close", "data": { "type": "amqp_close", "connection_id": "uuid" } }