FlowX

Generate, run, and evolve AI agent workflows from natural language

English · 简体中文

⚙️ Setup · 🚀 Quick Start · ✨ Key Features · 🧰 MCP Tools · 🐍 Python SDK · 🤝 A2A · 🏗️ Architecture · 📖 Docs · 简体中文

FlowX lets AI agents build, run, and evolve workflows through MCP, a Python SDK, or A2A. Describe a process and FlowX generates a workflow graph, audited and tested Python nodes, and a FastAPI backend.

Describe → Generate → Inspect → Run → Evolve → Reuse

Original upstream: AIpRoBuilder/FlowX · Attribution notice · Source provenance and verification. FlowX remains Apache-2.0 licensed; compliant forks and reuse are welcome.

🌟 Why FlowX?

Turn an intent into an editable, reusable process instead of a one-off answer. Change the workflow later without rebuilding unchanged nodes.

Illustrative walkthrough, not a live recording. The UI, cards, and timings are illustrative; FlowX does not bundle this interface. See the storyboard.

🚀 Key Features

• 🧩 Generate a graph, node documentation, Python code, and a backend.
• 🛡️ Audit and test each node, with bounded repair; failing nodes are not published.
• 🔁 Evolve the workflow or individual nodes, reusing unchanged nodes.
• 🚀 Run locally through MCP or the SDK; MCP also supports step inputs and session resets.
• 🔌 Choose an interface: local stdio MCP, Python SDK, or non-streaming A2A.
• 📦 Reuse artifacts stored under workspace/workflows/<workflow_id>.
• 🧠 Bring your own model and UI: the optional DeepSeek Harness plugin is planned to provide a read-only workflow viewer; use the separate FlowX skill to create, evolve, and run workflows.

⚙️ Setup

Prerequisites

• Python 3.10 or later.
• Git, to obtain FlowX and its workflow runtime dependency.
• An MCP-capable agent client for the local stdio path.
• An OpenAI-compatible LLM provider (DeepSeek is the default).

Installation

Clone FlowX and install the core, MCP server, SDK, and A2A adapter:

pip

git clone https://github.com/AIpRoBuilder/FlowX.git
cd FlowX

# Create and activate a virtual environment
python3.10 -m venv .venv
source .venv/bin/activate

# Install FlowX and its runtime dependencies
python -m pip install --upgrade pip
python -m pip install -e .

# Poetry
poetry env use python3.10
poetry install

# uv
uv sync

# conda
conda env create -f environment.yml
conda activate flowx

python3.10 -m venv .venv
source .venv/bin/activate
python -m pip install --upgrade pip
python -m pip install -e .
python -m pip install -e ../ag_ui_worflow --no-deps

# Poetry
poetry env use python3.10
poetry install
poetry run pip install -e ../ag_ui_worflow --no-deps

# uv
uv sync
uv pip install --python .venv/bin/python -e ../ag_ui_worflow --no-deps

For source-checkout execution, FlowX automatically discovers each packages/* component. Set FLOWX_EXTRA_PATHS only when the external ag_ui_workflow source is not installed.

Configure the environment

cp .env.example .env

Edit .env and set your provider, model, API key, and workspace:

FLOWX_LLM_PROVIDER=deepseek
FLOWX_LLM_MODEL=deepseek-flash
FLOWX_LLM_API_KEY=
FLOWX_DEFAULT_WORKSPACE=./.flowx_workspaces

FLOWX_CONFIG_ROOT lets an installed flowx-mcp resolve a .env file outside the repository checkout. FLOWX_EXTRA_PATHS can point at a local ag_ui_workflow source when it is not installed into the current environment.

The SDK and A2A server read process environment variables; they do not automatically load this .env file. Export the settings or load the file in your application before creating a client. For example, from the checkout root in macOS/Linux:

set -a; source .env; set +a

Start the MCP server

# Default local stdio entry point
flowx-mcp

# Verbose logging
flowx-mcp --verbose

# Source-checkout entry point
python3.10 run_server.py

Use poetry run or uv run if the environment is not activated. An MCP host normally starts the server itself; the command above is for checking the entry point.

Register FlowX in an MCP client

Replace the executable with the absolute path to your virtual environment's flowx-mcp, and set your credentials and workspace. MCP hosts may not inherit your shell's PATH.

{
  "mcpServers": {
    "flowx": {
      "command": "/absolute/path/to/your/venv/bin/flowx-mcp",
      "env": {
        "FLOWX_LLM_PROVIDER": "deepseek",
        "FLOWX_LLM_MODEL": "deepseek-flash",
        "FLOWX_LLM_API_KEY": "<your key>",
        "FLOWX_DEFAULT_WORKSPACE": "/absolute/path/to/flowx_workspaces"
      }
    }
  }
}

Store credentials according to your MCP client's security guidance; avoid committing them to a repository.

Install the FlowX agent skill

TARGET_PROJECT=/absolute/path/to/your/project
mkdir -p "$TARGET_PROJECT/.github/skills/flowx-workflows"
cp -R skills/flowx-workflows/. "$TARGET_PROJECT/.github/skills/flowx-workflows/"

mkdir -p ~/.copilot/skills/flowx-workflows
cp -R skills/flowx-workflows/. ~/.copilot/skills/flowx-workflows/

hermes skills install AIpRoBuilder/FlowX/skills/flowx-workflows
hermes skills list

python3 -c 'import sys; assert sys.version_info >= (3, 10), "Python 3.10+ required"' && \
  test ! -e "$HOME/.flowx/venv" && \
  test ! -L "$HOME/.flowx/venv" && \
  python3 -m venv "$HOME/.flowx/venv" && \
  "$HOME/.flowx/venv/bin/python" -m pip install 'git+https://github.com/AIpRoBuilder/FlowX.git' && \
  hermes skills install AIpRoBuilder/FlowX/skills/flowx-workflows

mkdir -p ~/.hermes/skills/flowx-workflows
cp -R skills/flowx-workflows/. ~/.hermes/skills/flowx-workflows/

⚡ Quick Start

After setup, ask your MCP-capable agent:

1. Describe a workflow

Use flowx to build a workflow.

Build a workflow that accepts a support ticket, classifies its category,
assigns a priority, and drafts a reply. Name the reply node draft_reply.
Return the category, priority, and draft reply as structured output.

2. Build and inspect

1. Call create_workflow(name="Ticket Triage"), then start_new_workflow(workflow_id, prompt). The optional name is the workflow folder and ID; omitting it keeps the generated-ID behavior. A name must be a safe single folder name and cannot overwrite an existing path.
2. Poll get_build_status(workflow_id). Check status and result.ok: the response's top-level ok only confirms that the status query succeeded.
3. If input_required, show awaiting.prompt and awaiting.test_log, then resume with submit_background_build_input using the exact workflow_id, revision, request_id, awaiting.stepId as step_id, and the user's input. Poll again.
4. Inspect the result with get_workflow and list_nodes. Use the generated step IDs, not fixed demo IDs.

MCP builds run in the background; the Python SDK retains synchronous methods. Older synchronous MCP build tools were removed: finish or abort an old paused build with the old server before upgrading. Generation sends requirements and possibly logs to your LLM provider and can run generated tests; review generated code before running its service.

3. Run, evolve, and reuse

Run the workflow as a service.

Update the draft_reply node: for urgent tickets, include an escalation note.
Then rerun the service.

run_service starts or restarts the generated service. run_step feeds one step's input to the running service and reports when the workflow pauses for more input; reset_session cancels a paused run. start_evolve_workflow changes the whole workflow (target="workflow"), one node (a step id or node id from list_nodes), or lets the engine decide; poll its background build as above.

🧰 MCP Tool Surface

These MCP tools let the host agent operate the workflow layer from the same conversation. UI rendering is deliberately left to the host.

🐍 Python SDK

Embed workflow building directly in a Python agent or the backend behind your own intelligent UI, without an MCP transport. A complete walkthrough is in examples/sdk_quickstart.py.

from flowx_sdk import FlowXClient

with FlowXClient(workspace="./workflows") as flowx:
    created = flowx.build_workflow_json(
        "Build a workflow that categorizes and prioritizes support tickets.",
    ).raise_for_failure()                                  # workflow.json + requirements
    workflow_id = created.workflow_id
    flowx.build_nodes(workflow_id, await_repair_input=False).raise_for_failure()  # all nodes, or {step_id: amendment}
    flowx.build_main_entrypoint(workflow_id).raise_for_failure()
    print(flowx.run_service(workflow_id))                  # run as a service
    input("Press Enter to stop the service... ")

Each build call returns an AgentResult with per-stage outcomes, errors, logs, and artifact paths. raise_for_failure() raises for failed stages. Keep the client open while using the service; exiting the with block stops services it started.

build_nodes, build_workflow, and evolve default to pausing on a failed node test. If status="input_required", inspect awaiting and resume with flowx.answer(request_id, input, awaiting["stepId"]). A paused result can have ok=True, so raise_for_failure() alone does not confirm completion. Pass await_repair_input=False to repair from the test log automatically. For guidance and input options, see examples/sdk_guided_repair.py.

Answer a node-build pause from the terminal

With the SDK installed, LLM settings exported, and a workflow graph already built, run the skill's build script from the repository root, replacing the workspace and workflow ID with your own:

python skills/flowx-workflows/scripts/build.py --workspace "$WS" nodes --workflow-id "$WF"

If the script emits an input_required JSON event, review awaiting.prompt and awaiting.test_log. Keep the same process running and enter one JSON line in its stdin, followed by Enter:

{"action":"answer","input":"Handle empty input before calling the API."}

Send one answer per pause; repeat if another node needs input. Use test: <guidance> as the input to amend the generated test, skip to skip this amendment attempt, or an empty string to repair from the test log alone. Send {"action":"abort"} to cancel; closing stdin also aborts. The optional step_id must match awaiting.stepId. --auto-repair disables these test-repair pauses. Check the final result.status, result.ok, stages, errors, and logs rather than assuming a pause or answer means success.

Node audit violations are repaired automatically from their audit reports; they do not have a separate user-input pause. See the skill instructions and dynamic workflow documentation for details.

To change an existing workflow, choose what to evolve with evolve(workflow_id, command, target=...). list_nodes returns each node's step id and node id to use as a target; without a target the engine decides what the command refers to. A runnable walkthrough is in examples/sdk_evolve.py.

For a direct, LLM-free smoke test of the expanded DynamicWorkflowOrchestrator engine, see examples/dynamic_workflow_orchestrator.py. It exercises per-node generation/repair chains, finalizer IDs and an incremental business-graph update.

from flowx_sdk import FlowXClient, WORKFLOW_TARGET

workflow_id = "<existing workflow ID>"
with FlowXClient(workspace="./workflows") as flowx:
  flowx.evolve(workflow_id, "Add a language-detection step.", target=WORKFLOW_TARGET, await_repair_input=False).raise_for_failure()
  nodes = flowx.list_nodes(workflow_id)
  flowx.evolve(workflow_id, "Validate empty input.", target=nodes[0]["step_id"], await_repair_input=False).raise_for_failure()

🤝 A2A Server

Let another agent delegate workflow generation to FlowX through the non-streaming Agent2Agent endpoint:

flowx-a2a --host 127.0.0.1 --port 8001

It serves an A2A agent card at /.well-known/agent-card.json and JSON-RPC at /a2a (also /). Supported methods are message/send, tasks/get, and tasks/cancel. Send an A2A text message to build a workflow; add metadata.workflowId to modify an existing one. The task is completed when every stage succeeded and otherwise failed; either way it carries the result, including the generated artifact paths.

🏗️ Architecture

The Generative Workflow layer has one shared core and three agent-facing interfaces: flowx_core plans, generates, audits, and runs the generated workflows; flowx_mcp, flowx_sdk, and flowx_a2a make those capabilities available to different agent hosts. The generated runtime sits beneath that layer; the intelligent UI sits above it.

flowchart TB
  UI["Intelligent UI / agent host"] --> MCP["Local MCP: build, modify, run"]
  UI --> SDK["Python SDK: build, modify, run"]
  PEER["Peer agent"] --> A2A["A2A: delegate generation"]
  MCP --> CORE["FlowX · Generative Workflow layer"]
  SDK --> CORE
  A2A --> CORE
  CORE --> FILES["Workflow graph + node code + backend"]
  FILES --> RUNTIME["FastAPI + ag_ui_workflow runtime"]
  MCP -->|"run / rerun as a service"| RUNTIME
  RUNTIME -.->|"AG-UI SSE to a custom frontend"| UI
  MCP -->|"follow-up changes"| CORE

Repository layout

Artifact flow

Workflow compilation is the mechanism behind the Generative Workflow layer. Planning and code generation turn intent into artifacts; audits provide feedback during generation and runtime checks. The artifact flow below is mirrored from assets/architecture.mmd.

flowchart LR
    A[Raw requirement] --> B[requirement_analysis.md]
    B --> C[workflow.json]
    C --> D[node_docs/*.md]
    D --> E[nodes/*.py]
    E --> F[main.py]

    C --> G[Graph audit]
    E --> H[Node audit]
    F --> I[Entrypoint audit]
    F --> J[Runtime log audit]

    G -->|feedback| C
    H -->|feedback| E
    I -->|feedback| F
    J -->|feedback| E

🧪 Runtime Compatibility

FlowX targets the request-scoped ag_ui_workflow API, verified against upstream revision 0319046 (2026-10-23), including the StepMeta step-metadata contract. Contract details for contributors are in docs/RUNTIME.md.

🙏 Acknowledgements

FlowX thanks CGraph for providing the runtime DAG engine framework used in workflow execution.

🤝 Contributing

Issues and pull requests are welcome. If FlowX is useful to you, star the repo, share a workflow you generated, or open an issue describing an intent you want turned into a workflow.

When you redistribute or modify FlowX, keep the LICENSE and NOTICE attribution (see docs/PROVENANCE.md).

💬 Contact

Interested in building the Generative Workflow layer for AI Agents—or connecting it to an intelligent UI? Get in touch at peterxcx@gmail.com.

WeChat QR code:

📜 License & Provenance

FlowX is licensed under the Apache License 2.0. Redistribution and modification are permitted under its terms; keep the NOTICE attribution and the source provenance record when redistributing FlowX source.

Original upstream: AIpRoBuilder/FlowX.

⭐ Star History

↑ Back to Top ↑