# flowmcp

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

## The problem

Generalist agents (declarative [kagent](https://kagent.dev), [deep agents](https://github.com/langchain-ai/deepagents),
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.

## How it does it

flowmcp is a small server that reads a directory of [Prompt flow](https://microsoft.github.io/promptflow)
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

```bash
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](#3-use-it-as-an-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`](https://docs.astral.sh/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`:

```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](#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](#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:

```bash
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`):

```bash
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:

```json
{
  "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):

```bash
# 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.

| Parameter | Required | Description |
|---|---|---|
| `url` | ✓ | Full request URL. |
| `method` | | HTTP method (default `GET`). |
| `headers` | | Request headers (auth, content-type, …). |
| `params` | | URL query parameters. |
| `json_body` | | Body serialized as JSON. |
| `data` | | Raw/form body (when not JSON). |
| `timeout` | | Seconds (default `30`). |
| `verify` | | Verify TLS (default `true`). |

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`.

| Parameter | Required | Description |
|---|---|---|
| `model` | ✓ | Model name, e.g. `nova-pro`. |
| `api_key` | ✓ | Bearer token for the endpoint. |
| `messages` | ✓ | OpenAI-style messages list. |
| `base_url` | | OpenAI-compatible base URL (default `https://api.openai.com/v1`); `/chat/completions` is appended. |
| `response_format` | | Predefined output format fed to the model. With `{"type": "json_object"}` or `{"type": "json_schema", ...}` the parsed object is returned instead of text. |
| `temperature` | | Sampling temperature (default `0.7`). |
| `max_tokens` | | Max completion tokens. |
| `extra_body` | | Provider-specific fields merged into the request. |

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.

| Parameter | Required | Description |
|---|---|---|
| `user` | ✓ | User message content. |
| `system` | | System instruction (prepended). |
| `context` | | Data (e.g. an API response) appended to the user message so the model can reason over it. |
| `history` | | Prior `{role, content}` messages inserted before the current turn. |

### `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`.

| Parameter | Required | Description |
|---|---|---|
| `template` | ✓ | Jinja2 template string. |
| `variables` | | Variables available to the template. |

### `parse_json` — extract / reshape structured data

`flowmcp.tools.parse_json.parse_json` — run a [JMESPath](https://jmespath.org)
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.

| Parameter | Required | Description |
|---|---|---|
| `data` | ✓ | Structured input (dict/list), **or a JSON string** (e.g. an LLM reply) — a JSON string, including one wrapped in a ```` ```json ```` fence, is parsed automatically. |
| `query` | ✓ | JMESPath expression. |
| `default` | | Returned when the query result is null/absent. |

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).

| Parameter | Required | Description |
|---|---|---|
| `a`–`d` | | Branch outputs; first non-empty is returned. |

---

## 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:

```yaml
- 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:
  ```yaml
  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:

```yaml
- 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:
> ```yaml
> 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

```bash
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`):

```bash
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](#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

```bash
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.