flowmcp

Give a generalist agent a deterministic, isolated, permission-bound operation — as a single MCP tool.

The problem

Generalist agents (declarative kagent, deep agents, Claude Code, Cursor, …) are powerful but non-deterministic and unbounded: they decide their own steps, and the more capability you hand them, the wider the blast radius. For many real tasks you don't want improvisation — you want a specific operation to run the same way every time, touching only the systems it's allowed to touch.

flowmcp lets you carve that operation out as a small, declarative workflow and expose it to any agent as one MCP tool:

• Deterministic — the workflow is a DAG you author in YAML. Given the same inputs it runs the same nodes in the same order. Any LLM reasoning inside is confined to steps you placed, and even branching is explicit (the LLM emits a structured decision; the flow acts on it).

• Isolated — each workflow is its own tool with its own derived input/output schema. No shared state, no ambient capability. Run one flow per pod / namespace for hard tenant isolation.

• Permission-bound — a flow can only do what its nodes' built-in tools allow (call this endpoint, talk to this LLM gateway). No arbitrary customer code runs in the pod; Kubernetes RBAC confines each release to its namespace.

• Trivially integrated — it speaks MCP, so any generalist agent discovers and calls it like any other tool. The agent stays generalist; the risky, must-be-repeatable part is delegated to a bounded workflow.

Related MCP server: skillsmcp

How it does it

flowmcp is a small server that reads a directory of Prompt flow workflows and exposes each one as an MCP tool. You write a flow.dag.yaml; flowmcp automatically derives the tool's input/output schema from it and serves it over MCP — no code to write, build, or host.

The reusable Python — like calling an LLM or a REST endpoint — already ships inside flowmcp as a package tool. Your flow references it by name, so you provide only YAML.

your flow.dag.yaml ──► flowmcp ──► MCP tool  ──► called by any generalist agent
   (you write this)     (server)   (auto-generated)      (kagent, deep agent, …)

Quickstart

uv sync
uv run flowmcp --flows-dir ./example-flows

This loads every flow under ./example-flows (the repo ships 7 sample flows, e.g. say-hi) and serves the MCP endpoint at http://127.0.0.1:8080/mcp. Jump to Using the MCP tool to call it.

example-flows/ holds runnable samples only — they are not baked into the Docker image. At runtime you supply your own flows dir (mounted at /flows).

Requirements: Python 3.11 (Prompt flow doesn't support 3.12+ cleanly) and uv.

Tutorial: create your own flow and use it as an MCP tool

We'll build a tool called summarize that asks an LLM to summarize text.

1. Author the flow

Create a directory under your flows dir — the directory name becomes the MCP tool name — and add a flow.dag.yaml. Nothing else: this flow is pure YAML because every step references a built-in package tool.

flows/
└── summarize/
    └── flow.dag.yaml

flows/summarize/flow.dag.yaml:

$schema: https://azuremlschemas.azureedge.net/promptflow/latest/Flow.schema.json

# The MCP tool description — this is what the calling agent's model reads to
# decide when to invoke the tool. Set it deliberately. If omitted, it falls
# back to "Run the '<flow-dir-name>' flow."
description: Summarize a piece of text into a single sentence.

# These become the MCP tool's parameters. An input WITHOUT a default is
# marked "required" in the tool schema; one WITH a default is optional.
inputs:
  text:
    type: string
  api_key:
    type: string
  model:
    type: string
    default: nova-pro
  base_url:
    type: string
    default: https://<llm-gateway-url>/v1

# These become the MCP tool's result (outputSchema).
outputs:
  summary:
    type: string
    reference: ${ask_llm.output}

nodes:
  # Build the chat messages. References flowmcp's BUILT-IN package tool —
  # no code shipped. `system` sets the instruction, `user` is the text to work on.
  - name: build_messages
    type: python
    source:
      type: package
      tool: flowmcp.tools.messages.build_messages
    inputs:
      system: Summarize the user's text in one sentence.
      user: ${inputs.text}

  # Call the LLM. This also references a BUILT-IN package tool by name —
  # no code shipped, no connection to configure.
  - name: ask_llm
    type: python
    source:
      type: package
      tool: flowmcp.tools.chat_completion.chat_completion
    inputs:
      model: ${inputs.model}
      api_key: ${inputs.api_key}
      base_url: ${inputs.base_url}
      messages: ${build_messages.output}

That's the whole workflow — one YAML file, no Python. Key ideas:

• Wiring is by reference. ${inputs.text} reads a tool argument; ${build_messages.output} feeds one node's result into the next. The reference graph is the execution order.

• Every step is a built-in. source.type: package + a tool: identifier uses a tool that ships inside flowmcp (build_messages, chat_completion, …) — you don't write or configure it. See Built-in tools.

• You can add more steps. Chain nodes, fan out in parallel (independent nodes run concurrently), or branch with activate conditions — see Building richer flows.

YAML tip: inside a flow.dag.yaml, write inputs/outputs in block style (as above). Inline maps like text: { type: string } are rejected by Prompt flow's parser.

2. (Optional) test the flow directly

Before serving it, you can run the flow on its own:

uv run python -c "
from flowmcp.runner import run_flow
from pathlib import Path
print(run_flow(Path('flows/summarize'),
      {'text': 'MCP lets agents call tools over a standard protocol.',
       'api_key': 'YOUR_TOKEN'}))
"

3. Use it as an MCP tool

Start the server (it auto-discovers summarize):

uv run flowmcp --flows-dir ./flows

You'll see it register the flow:

INFO flowmcp.loader: Registered flow 'summarize' from .../flows/summarize/flow.dag.yaml
INFO flowmcp: Loaded 1 flow(s): summarize

Connect an MCP client (e.g. Claude Code / Cursor) — point it at the endpoint:

{
  "mcpServers": {
    "flowmcp": {
      "command": "npx",
      "args": ["-y", "mcp-remote", "http://localhost:8080/mcp", "--transport", "http-only"]
    }
  }
}

The client now sees a summarize tool with parameters text (required), api_key (required), model, base_url, and a summary result — all derived from your YAML.

Or call it directly over HTTP (MCP streamable-HTTP handshake):

# 1. initialize — capture the mcp-session-id response header
SID=$(curl -s -D - -o /dev/null -X POST http://localhost:8080/mcp \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2024-11-05","capabilities":{},"clientInfo":{"name":"c","version":"1"}}}' \
  | awk -F': ' 'tolower($1)=="mcp-session-id"{print $2}' | tr -d '\r')

# 2. list tools
curl -s -X POST http://localhost:8080/mcp -H "mcp-session-id: $SID" \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":2,"method":"tools/list","params":{}}'

# 3. call the summarize tool
curl -s -X POST http://localhost:8080/mcp -H "mcp-session-id: $SID" \
  -H 'Content-Type: application/json' -H 'Accept: application/json, text/event-stream' \
  -d '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"summarize",
       "arguments":{"text":"MCP lets agents call tools over a standard protocol.",
                    "api_key":"YOUR_TOKEN"}}}'

The call runs your flow end to end and returns the summary in the tool result.

Add more tools by dropping more flow directories in — each becomes its own MCP tool. To serve several at once:

flows/
├── summarize/flow.dag.yaml
├── classify/flow.dag.yaml
└── translate/flow.dag.yaml

Built-in tools

These ship inside flowmcp. Reference any of them from a flow node via source: {type: package, tool: <identifier>} — no custom Python needed. The common agentic pattern — call an external system, then have an LLM reason over the result — is fully covered (see the fetch-and-reason example flow).

http_request — call any REST endpoint

flowmcp.tools.http_request.http_request — completely flexible: method, URL, headers, params, and body are all parameters.

Returns {status_code, ok, headers, json, text}. It does not raise on HTTP error statuses — inspect status_code/ok to branch or hand the error to an LLM.

chat_completion — call an LLM

flowmcp.tools.chat_completion.chat_completion — OpenAI-compatible /chat/completions.

Designed to be called repeatedly — a flow can talk to the LLM in several steps (e.g. draft → critique → rewrite).

build_messages — assemble chat messages

flowmcp.tools.messages.build_messages — build the messages list for chat_completion declaratively.

render_template — dynamic strings

flowmcp.tools.template.render_template — render a Jinja2 template from variables. Use for dynamic URLs, bodies, or prompts, e.g. https://api/users/{{ id }}. Output is always a string — for structured output, use parse_json.

parse_json — extract / reshape structured data

flowmcp.tools.parse_json.parse_json — run a JMESPath query against structured data and return structured output (value, list, or a newly-shaped object) — so you can feed it to another structured input (e.g. an http_request json_body) without custom code.

JMESPath examples: user.name (nested value), items[].id (project a list), items[?price > `10`].name (filter), {name: user.name, n: length(items)} (reshape into a new object). See the fetch-reshape-reason example flow.

render_template vs parse_json: template → string (prompts, URLs); parse_json → structured (objects/lists to pass onward).

coalesce — merge branches

flowmcp.tools.coalesce.coalesce — returns the first non-empty argument. Used as the merge node after an activate if/else: wire each branch's output to a parameter, and it returns whichever branch actually ran (the bypassed one is empty).

Shaping a prompt from selected response fields (Jinja2)

To build a prompt from specific fields of a REST response — or after transforming them — insert a render_template step between the HTTP call and the LLM. The whole response is passed in as a variable; the Jinja2 template picks and formats only what you need:

- name: call_api
  # ... http_request returns e.g. {"title": "...", "completed": false, "userId": 1}

- name: shape_prompt
  type: python
  source:
    type: package
    tool: flowmcp.tools.template.render_template
  inputs:
    variables:
      resp: ${call_api.output.json}       # whole response available as `resp`
    template: |-
      Title: {{ resp.title }}
      Status: {% if resp.completed %}DONE{% else %}NOT DONE{% endif %}
      Owner: user #{{ resp.userId }}
      In one sentence, does this need attention?

- name: reason
  # ... chat_completion with messages built from ${shape_prompt.output}

Jinja2 gives you field selection (resp.user.name), filters/transforms ({{ resp.tags | join(", ") }}, | upper, | round, | default("n/a")), and conditionals/loops ({% if %}, {% for %}) — so "a few fields, after some modification" needs no custom code. See the fetch-extract-reason example flow.

Gotcha: a response key that collides with a Python dict method (e.g. items, keys, values) must use bracket access — {{ resp['items'] }}, not {{ resp.items }}.

Building richer flows

Prompt flow evaluates nodes as a DAG, so you can compose:

• Sequential steps — chain nodes via ${node.output} references.

• Parallel steps — nodes with no dependency between them run concurrently; a later node that references several of them fans the results back in.

• Conditional branches (if/else, switch) — add an activate block so a node runs only when a condition holds; the other branch's nodes are bypassed:

  activate:
    when: ${call_api.output.ok}   # branch on a prior node's output
    is: true

  Give each branch the opposite condition (is: true / is: false), then merge with the coalesce tool — it returns whichever branch produced output (the bypassed one resolves to empty). See the fetch-branch-reason example. This is execution branching (which nodes run), distinct from Jinja2 {% if %} which only branches text.

The graph is acyclic — great for deterministic pipelines, but it cannot loop. For bounded iteration, loop inside a single python step.

Deterministic-agentic pattern: the LLM decides, the flow acts

The keystone of a deterministic agentic workflow: the LLM produces a structured decision, and the flow branches on it deterministically. Ask for JSON in the prompt, parse it, then activate on the extracted value:

- name: decide            # chat_completion — prompt asks for {"approved": bool, "reason": "..."}
- name: approved          # parse_json: data=${decide.output}, query="approved"
- name: on_approve        # activate: when ${approved.output} is true
- name: on_reject         # activate: when ${approved.output} is false
- name: route             # coalesce the branches

Two robustness notes baked into parse_json, learned from real gateway behavior:

• Ask for JSON in the prompt, not via response_format. The inference gateway rejects the OpenAI response_format field, so instruct the model (e.g. "Reply as JSON: {...}") and let parse_json parse the reply.

• parse_json handles JSON returned as a string, including ```json markdown fences — so chat_completion text output can be queried directly.

See the llm-decision (binary approve/reject) and classify-route (N-way switch) example flows — both verified end to end.

Reference-resolution rule (important): a ${...} reference resolves only when it is the entire value of a node input — e.g. variables: ${call_api.output.json}. A reference nested inside a dict/map literal is not resolved and stays literal:

variables:                       # ❌ WRONG — ${...} stays literal
  resp: ${call_api.output.json}
variables: ${call_api.output.json}   # ✅ pass the whole object; use its keys in the template

Adding a new built-in tool (for maintainers)

Customers write only YAML; new capabilities are added here as package tools:

1. Write a @tool-decorated function in src/flowmcp/tools/.

2. Add its metadata to src/flowmcp/tools/_meta.py (identifier → function/module

   • input types).

The package_tools entry point (in pyproject.toml) exposes it automatically, so any flow can reference it by its module.function identifier. No other wiring needed.

Deployment

Docker

docker build -t flowmcp .
docker run -p 8080:8080 -v "$PWD/example-flows:/flows" flowmcp

Serves 0.0.0.0:8080/mcp and reads flows from /flows (mount a volume or ConfigMap). Configure via env vars: FLOWMCP_FLOWS_DIR, FLOWMCP_HOST, FLOWMCP_PORT, FLOWMCP_TRANSPORT (default streamable-http).

Kubernetes (Helm)

The chart in chart/ deploys a Deployment + Service and supplies flows inline via values (rendered into a ConfigMap, mounted at /flows):

helm install flowmcp ./chart -f ./chart/values-example.yaml

Define flows under flows.<name> — each value is the flow's flow.dag.yaml content (a YAML string) and each <name> becomes an MCP tool. values-example.yaml shows a flow. Changing flows rolls the pods automatically. For per-tenant isolation, install one release per namespace with a namespace-scoped ServiceAccount; standard Kubernetes RBAC then confines each pod to that namespace's resources.

YAML-only by design. The chart accepts declarative YAML flows only — a flow value must be a YAML string, never a map of files. Supplying a .py fails the render with a clear error. All reusable logic ships in the image as built-in tools; flows reference them via source: {type: package, tool: flowmcp.tools.<...>}. This keeps arbitrary customer code out of the pod (important for multi-tenant/RBAC safety).

How it works

flows/<name>/flow.dag.yaml ──► loader ──► FlowRegistry ──► MCP server (streamable-HTTP)
                               (schema)     (dynamic)         one tool per flow
                                                              call = run the flow

• tools/ — built-in package tools (via the package_tools entry point).

• registry.py — dynamic register/unregister/list; the stable core.

• loader.py — discovers flow.dag.yaml files and registers them.

• schema.py — flow inputs/outputs → MCP JSON schema (required = no default).

• runner.py — runs a flow in-process via load_flow.

• server.py — MCP server; list_tools reads the registry live, call_tool runs the flow. No authentication.

Tests

uv run pytest

Roadmap

Phase 2 — a Kubernetes controller reconciling a Workflow custom resource into FlowRegistry.register/unregister. Because ingest is decoupled from the registry, the MCP server and tools stay unchanged; the controller is just a second source feeding the same API.