flowstudio-power-automate-mcp

Compare original and translation side by side

🇺🇸

Original

English
🇨🇳

Translation

Chinese

Power Automate via FlowStudio MCP

通过FlowStudio MCP使用Power Automate

This skill lets AI agents read, monitor, and operate Microsoft Power Automate cloud flows programmatically through a FlowStudio MCP server — no browser, no UI, no manual steps.
Requires: A FlowStudio MCP subscription (or compatible Power Automate MCP server). You will need:
  • MCP endpoint:
    https://mcp.flowstudio.app/mcp
    (same for all subscribers)
  • API key / JWT token (
    x-api-key
    header — NOT Bearer)
  • Power Platform environment name (e.g.
    Default-<tenant-guid>
    )

该技能可让AI Agent通过FlowStudio MCP服务器以编程方式读取、监控和操作Microsoft Power Automate云流,无需浏览器、无需操作UI、也无需手动步骤。
要求: 需拥有FlowStudio MCP订阅(或兼容的Power Automate MCP服务器)。你需要准备:
  • MCP端点:
    https://mcp.flowstudio.app/mcp
    (所有订阅用户通用)
  • API密钥 / JWT令牌(放在
    x-api-key
    请求头中,不是Bearer认证)
  • Power Platform环境名称(例如
    Default-<租户GUID>

Source of Truth

可信信息来源

PrioritySourceCovers
1Real API responseAlways trust what the server actually returns
2
tools/list
Tool names, parameter names, types, required flags
3SKILL docs & reference filesResponse shapes, behavioral notes, workflow recipes
Start every new session with
tools/list
.
It returns the authoritative, up-to-date schema for every tool — parameter names, types, and required flags. The SKILL docs cover what
tools/list
cannot tell you: response shapes, non-obvious behaviors, and end-to-end workflow patterns.
If any documentation disagrees with
tools/list
or a real API response, the API wins.

优先级来源覆盖范围
1实际API响应始终以服务器实际返回的结果为准
2
tools/list
工具名称、参数名称、参数类型、必填标识
3技能文档与参考文件响应结构、行为说明、工作流示例
每次新会话都请先调用
tools/list
它会返回所有工具的权威、最新的schema定义,包括参数名称、类型和必填标识。技能文档会补充
tools/list
无法提供的信息:响应结构、非显性行为、端到端工作流模式。
如果任何文档内容与
tools/list
返回结果或实际API响应冲突,以API结果为准。

Recommended Language: Python or Node.js

推荐使用语言:Python或Node.js

All examples in this skill and the companion build / debug skills use Python with
urllib.request
(stdlib — no
pip install
needed). Node.js is an equally valid choice:
fetch
is built-in from Node 18+, JSON handling is native, and the async/await model maps cleanly onto the request-response pattern of MCP tool calls — making it a natural fit for teams already working in a JavaScript/TypeScript stack.
LanguageVerdictNotes
Python✅ RecommendedClean JSON handling, no escaping issues, all skill examples use it
Node.js (≥ 18)✅ RecommendedNative
fetch
+
JSON.stringify
/
JSON.parse
; async/await fits MCP call patterns well; no extra packages needed
PowerShell⚠️ Avoid for flow operations
ConvertTo-Json -Depth
silently truncates nested definitions; quoting and escaping break complex payloads. Acceptable for a quick
tools/list
discovery call but not for building or updating flows.
cURL / Bash⚠️ Possible but fragileShell-escaping nested JSON is error-prone; no native JSON parser
TL;DR — use the Core MCP Helper (Python or Node.js) below. Both handle JSON-RPC framing, auth, and response parsing in a single reusable function.

本技能及配套的构建/调试技能的所有示例均使用Python的
urllib.request
(标准库,无需
pip install
)。Node.js也是完全合适的选择:Node 18+内置
fetch
,原生支持JSON处理,async/await模型与MCP工具调用的请求-响应模式匹配度很高,非常适合已经使用JavaScript/TypeScript技术栈的团队。
语言推荐度说明
Python✅ 推荐简洁的JSON处理,无转义问题,所有技能示例均使用该语言
Node.js (≥ 18)✅ 推荐原生支持
fetch
+
JSON.stringify
/
JSON.parse
;async/await非常适配MCP调用模式;无需额外安装依赖包
PowerShell⚠️ 不建议用于流操作
ConvertTo-Json -Depth
会静默截断嵌套定义;引号和转义会导致复杂 payload 出错。仅可用于快速调用
tools/list
做探索,不适合构建或更新流
cURL / Bash⚠️ 可实现但稳定性差Shell转义嵌套JSON极易出错;无原生JSON解析器
总结:使用下方的核心MCP辅助函数(Python或Node.js版本)。 两者都在单个可复用函数中实现了JSON-RPC帧处理、鉴权和响应解析。

What You Can Do

支持的操作

FlowStudio MCP has two access tiers. FlowStudio for Teams subscribers get both the fast Azure-table store (cached snapshot data + governance metadata) and full live Power Automate API access. MCP-only subscribers get the live tools — more than enough to build, debug, and operate flows.
FlowStudio MCP有两个访问层级。FlowStudio for Teams订阅用户可同时使用快速Azure表存储(缓存快照数据+治理元数据)和完整的Power Automate实时API访问权限。仅MCP订阅用户可使用实时工具,完全满足构建、调试和操作流的需求。

Live Tools — Available to All MCP Subscribers

实时工具 — 所有MCP订阅用户可用

ToolWhat it does
list_live_flows
List flows in an environment directly from the PA API (always current)
list_live_environments
List all Power Platform environments visible to the service account
list_live_connections
List all connections in an environment from the PA API
get_live_flow
Fetch the complete flow definition (triggers, actions, parameters)
get_live_flow_http_schema
Inspect the JSON body schema and response schemas of an HTTP-triggered flow
get_live_flow_trigger_url
Get the current signed callback URL for an HTTP-triggered flow
trigger_live_flow
POST to an HTTP-triggered flow's callback URL (AAD auth handled automatically)
update_live_flow
Create a new flow or patch an existing definition in one call
add_live_flow_to_solution
Migrate a non-solution flow into a solution
get_live_flow_runs
List recent run history with status, start/end times, and errors
get_live_flow_run_error
Get structured error details (per-action) for a failed run
get_live_flow_run_action_outputs
Inspect inputs/outputs of any action (or every foreach iteration) in a run
resubmit_live_flow_run
Re-run a failed or cancelled run using its original trigger payload
cancel_live_flow_run
Cancel a currently running flow execution
工具功能
list_live_flows
直接调用PA API列出环境中的流(数据始终是最新的)
list_live_environments
列出服务账号可见的所有Power Platform环境
list_live_connections
调用PA API列出环境中的所有连接
get_live_flow
获取完整的流定义(触发器、操作、参数)
get_live_flow_http_schema
查看HTTP触发流的JSON请求体结构和响应结构
get_live_flow_trigger_url
获取HTTP触发流的当前已签名回调URL
trigger_live_flow
向HTTP触发流的回调URL发送POST请求(自动处理AAD鉴权)
update_live_flow
单次调用即可创建新流或更新现有流定义
add_live_flow_to_solution
将非解决方案流迁移到解决方案中
get_live_flow_runs
列出近期运行历史,包含状态、起止时间和错误信息
get_live_flow_run_error
获取失败运行的结构化错误详情(按操作拆分)
get_live_flow_run_action_outputs
查看运行中任意操作(或每个foreach迭代)的输入/输出
resubmit_live_flow_run
使用原始触发器payload重新运行失败或已取消的任务
cancel_live_flow_run
取消当前正在运行的流执行任务

Store Tools — FlowStudio for Teams Subscribers Only

存储工具 — 仅FlowStudio for Teams订阅用户可用

These tools read from (and write to) the FlowStudio Azure table — a monitored snapshot of your tenant's flows enriched with governance metadata and run statistics.
ToolWhat it does
list_store_flows
Search flows from the cache with governance flags, run failure rates, and owner metadata
get_store_flow
Get full cached details for a single flow including run stats and governance fields
get_store_flow_trigger_url
Get the trigger URL from the cache (instant, no PA API call)
get_store_flow_runs
Cached run history for the last N days with duration and remediation hints
get_store_flow_errors
Cached failed-only runs with failed action names and remediation hints
get_store_flow_summary
Aggregated stats: success rate, failure count, avg/max duration
set_store_flow_state
Start or stop a flow via the PA API and sync the result back to the store
update_store_flow
Update governance metadata (description, tags, monitor flag, notification rules, business impact)
list_store_environments
List all environments from the cache
list_store_makers
List all makers (citizen developers) from the cache
get_store_maker
Get a maker's flow/app counts and account status
list_store_power_apps
List all Power Apps canvas apps from the cache
list_store_connections
List all Power Platform connections from the cache

这些工具读写FlowStudio Azure表,表中存储了租户流的监控快照,补充了治理元数据和运行统计信息。
工具功能
list_store_flows
从缓存中搜索流,包含治理标识、运行失败率和所有者元数据
get_store_flow
获取单个流的完整缓存详情,包括运行统计和治理字段
get_store_flow_trigger_url
从缓存中获取触发器URL(即时返回,无需调用PA API)
get_store_flow_runs
最近N天的缓存运行历史,包含耗时和修复建议
get_store_flow_errors
仅缓存的失败运行记录,包含失败操作名称和修复建议
get_store_flow_summary
聚合统计数据:成功率、失败次数、平均/最大耗时
set_store_flow_state
通过PA API启动或停止流,并将结果同步回存储
update_store_flow
更新治理元数据(描述、标签、监控标识、通知规则、业务影响)
list_store_environments
从缓存中列出所有环境
list_store_makers
从缓存中列出所有制作者(公民开发者)
get_store_maker
获取制作者的流/应用数量和账号状态
list_store_power_apps
从缓存中列出所有Power Apps画布应用
list_store_connections
从缓存中列出所有Power Platform连接

Which Tool Tier to Call First

优先调用的工具层级

TaskToolNotes
List flows
list_live_flows
Always current — calls PA API directly
Read a definition
get_live_flow
Always fetched live — not cached
Debug a failure
get_live_flow_runs
get_live_flow_run_error
Use live run data
⚠️
list_live_flows
returns a wrapper object
with a
flows
array — access via
result["flows"]
.
Store tools (
list_store_flows
,
get_store_flow
, etc.) are available to FlowStudio for Teams subscribers and provide cached governance metadata. Use live tools when in doubt — they work for all subscription tiers.

任务工具说明
列出流
list_live_flows
数据始终最新,直接调用PA API
读取流定义
get_live_flow
始终实时获取,无缓存
调试失败任务
get_live_flow_runs
get_live_flow_run_error
使用实时运行数据
⚠️
list_live_flows
返回的是封装对象
,包含
flows
数组,通过
result["flows"]
访问。
存储工具(
list_store_flows
get_store_flow
等)仅对FlowStudio for Teams订阅用户开放,提供缓存的治理元数据。有疑问时请使用实时工具,所有订阅层级都可使用。

Step 0 — Discover Available Tools

步骤0 — 探索可用工具

Always start by calling
tools/list
to confirm the server is reachable and see exactly which tool names are available (names may vary by server version):
python
import json, urllib.request

TOKEN = "<YOUR_JWT_TOKEN>"
MCP   = "https://mcp.flowstudio.app/mcp"

def mcp_raw(method, params=None, cid=1):
    payload = {"jsonrpc": "2.0", "method": method, "id": cid}
    if params:
        payload["params"] = params
    req = urllib.request.Request(MCP, data=json.dumps(payload).encode(),
        headers={"x-api-key": TOKEN, "Content-Type": "application/json",
                 "User-Agent": "FlowStudio-MCP/1.0"})
    try:
        resp = urllib.request.urlopen(req, timeout=30)
    except urllib.error.HTTPError as e:
        raise RuntimeError(f"MCP HTTP {e.code} — check token and endpoint") from e
    return json.loads(resp.read())

raw = mcp_raw("tools/list")
if "error" in raw:
    print("ERROR:", raw["error"]); raise SystemExit(1)
for t in raw["result"]["tools"]:
    print(t["name"], "—", t["description"][:60])

始终先调用
tools/list
确认服务器可访问,并查看确切的可用工具名称(名称可能因服务器版本不同有差异):
python
import json, urllib.request

TOKEN = "<YOUR_JWT_TOKEN>"
MCP   = "https://mcp.flowstudio.app/mcp"

def mcp_raw(method, params=None, cid=1):
    payload = {"jsonrpc": "2.0", "method": method, "id": cid}
    if params:
        payload["params"] = params
    req = urllib.request.Request(MCP, data=json.dumps(payload).encode(),
        headers={"x-api-key": TOKEN, "Content-Type": "application/json",
                 "User-Agent": "FlowStudio-MCP/1.0"})
    try:
        resp = urllib.request.urlopen(req, timeout=30)
    except urllib.error.HTTPError as e:
        raise RuntimeError(f"MCP HTTP {e.code} — check token and endpoint") from e
    return json.loads(resp.read())

raw = mcp_raw("tools/list")
if "error" in raw:
    print("ERROR:", raw["error"]); raise SystemExit(1)
for t in raw["result"]["tools"]:
    print(t["name"], "—", t["description"][:60])

Core MCP Helper (Python)

核心MCP辅助函数(Python)

Use this helper throughout all subsequent operations:
python
import json, urllib.request

TOKEN = "<YOUR_JWT_TOKEN>"
MCP   = "https://mcp.flowstudio.app/mcp"

def mcp(tool, args, cid=1):
    payload = {"jsonrpc": "2.0", "method": "tools/call", "id": cid,
               "params": {"name": tool, "arguments": args}}
    req = urllib.request.Request(MCP, data=json.dumps(payload).encode(),
        headers={"x-api-key": TOKEN, "Content-Type": "application/json",
                 "User-Agent": "FlowStudio-MCP/1.0"})
    try:
        resp = urllib.request.urlopen(req, timeout=120)
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"MCP HTTP {e.code}: {body[:200]}") from e
    raw = json.loads(resp.read())
    if "error" in raw:
        raise RuntimeError(f"MCP error: {json.dumps(raw['error'])}")
    text = raw["result"]["content"][0]["text"]
    return json.loads(text)
Common auth errors:
  • HTTP 401/403 → token is missing, expired, or malformed. Get a fresh JWT from mcp.flowstudio.app.
  • HTTP 400 → malformed JSON-RPC payload. Check
    Content-Type: application/json
    and body structure.
  • MCP error: {"code": -32602, ...}
    → wrong or missing tool arguments.

后续所有操作都可使用该辅助函数:
python
import json, urllib.request

TOKEN = "<YOUR_JWT_TOKEN>"
MCP   = "https://mcp.flowstudio.app/mcp"

def mcp(tool, args, cid=1):
    payload = {"jsonrpc": "2.0", "method": "tools/call", "id": cid,
               "params": {"name": tool, "arguments": args}}
    req = urllib.request.Request(MCP, data=json.dumps(payload).encode(),
        headers={"x-api-key": TOKEN, "Content-Type": "application/json",
                 "User-Agent": "FlowStudio-MCP/1.0"})
    try:
        resp = urllib.request.urlopen(req, timeout=120)
    except urllib.error.HTTPError as e:
        body = e.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"MCP HTTP {e.code}: {body[:200]}") from e
    raw = json.loads(resp.read())
    if "error" in raw:
        raise RuntimeError(f"MCP error: {json.dumps(raw['error'])}")
    text = raw["result"]["content"][0]["text"]
    return json.loads(text)
常见鉴权错误:
  • HTTP 401/403 → 令牌缺失、过期或格式错误。从mcp.flowstudio.app获取新的JWT即可。
  • HTTP 400 → JSON-RPC payload格式错误。检查
    Content-Type: application/json
    和请求体结构。
  • MCP error: {"code": -32602, ...}
    → 工具参数错误或缺失。

Core MCP Helper (Node.js)

核心MCP辅助函数(Node.js)

Equivalent helper for Node.js 18+ (built-in
fetch
— no packages required):
js
const TOKEN = "<YOUR_JWT_TOKEN>";
const MCP   = "https://mcp.flowstudio.app/mcp";

async function mcp(tool, args, cid = 1) {
  const payload = {
    jsonrpc: "2.0",
    method: "tools/call",
    id: cid,
    params: { name: tool, arguments: args },
  };
  const res = await fetch(MCP, {
    method: "POST",
    headers: {
      "x-api-key": TOKEN,
      "Content-Type": "application/json",
      "User-Agent": "FlowStudio-MCP/1.0",
    },
    body: JSON.stringify(payload),
  });
  if (!res.ok) {
    const body = await res.text();
    throw new Error(`MCP HTTP ${res.status}: ${body.slice(0, 200)}`);
  }
  const raw = await res.json();
  if (raw.error) throw new Error(`MCP error: ${JSON.stringify(raw.error)}`);
  return JSON.parse(raw.result.content[0].text);
}
Requires Node.js 18+. For older Node, replace
fetch
with
https.request
from the stdlib or install
node-fetch
.

适用于Node.js 18+的等效辅助函数(内置
fetch
,无需安装依赖包):
js
const TOKEN = "<YOUR_JWT_TOKEN>";
const MCP   = "https://mcp.flowstudio.app/mcp";

async function mcp(tool, args, cid = 1) {
  const payload = {
    jsonrpc: "2.0",
    method: "tools/call",
    id: cid,
    params: { name: tool, arguments: args },
  };
  const res = await fetch(MCP, {
    method: "POST",
    headers: {
      "x-api-key": TOKEN,
      "Content-Type": "application/json",
      "User-Agent": "FlowStudio-MCP/1.0",
    },
    body: JSON.stringify(payload),
  });
  if (!res.ok) {
    const body = await res.text();
    throw new Error(`MCP HTTP ${res.status}: ${body.slice(0, 200)}`);
  }
  const raw = await res.json();
  if (raw.error) throw new Error(`MCP error: ${JSON.stringify(raw.error)}`);
  return JSON.parse(raw.result.content[0].text);
}
要求Node.js 18+。旧版本Node可将
fetch
替换为标准库的
https.request
,或安装
node-fetch
包。

List Flows

列出流

python
ENV = "Default-<tenant-guid>"

result = mcp("list_live_flows", {"environmentName": ENV})
python
ENV = "Default-<tenant-guid>"

result = mcp("list_live_flows", {"environmentName": ENV})

Returns wrapper object:

返回封装对象:

{"mode": "owner", "flows": [{"id": "0757041a-...", "displayName": "My Flow",

{"mode": "owner", "flows": [{"id": "0757041a-...", "displayName": "My Flow",

"state": "Started", "triggerType": "Request", ...}], "totalCount": 42, "error": null}

"state": "Started", "triggerType": "Request", ...}], "totalCount": 42, "error": null}

for f in result["flows"]: FLOW_ID = f["id"] # plain UUID — use directly as flowName print(FLOW_ID, "|", f["displayName"], "|", f["state"])

---
for f in result["flows"]: FLOW_ID = f["id"] # 纯UUID,可直接作为flowName使用 print(FLOW_ID, "|", f["displayName"], "|", f["state"])

---

Read a Flow Definition

读取流定义

python
FLOW = "<flow-uuid>"

flow = mcp("get_live_flow", {"environmentName": ENV, "flowName": FLOW})
python
FLOW = "<flow-uuid>"

flow = mcp("get_live_flow", {"environmentName": ENV, "flowName": FLOW})

Display name and state

显示名称和状态

print(flow["properties"]["displayName"]) print(flow["properties"]["state"])
print(flow["properties"]["displayName"]) print(flow["properties"]["state"])

List all action names

列出所有操作名称

actions = flow["properties"]["definition"]["actions"] print("Actions:", list(actions.keys()))
actions = flow["properties"]["definition"]["actions"] print("Actions:", list(actions.keys()))

Inspect one action's expression

查看单个操作的表达式

print(actions["Compose_Filter"]["inputs"])

---
print(actions["Compose_Filter"]["inputs"])

---

Check Run History

检查运行历史

python
undefined
python
undefined

Most recent runs (newest first)

最近的运行记录(按时间倒序)

runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW, "top": 5})
runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW, "top": 5})

Returns direct array:

返回直接数组:

[{"name": "08584296068667933411438594643CU15",

[{"name": "08584296068667933411438594643CU15",

"status": "Failed",

"status": "Failed",

"startTime": "2026-02-25T06:13:38.6910688Z",

"startTime": "2026-02-25T06:13:38.6910688Z",

"endTime": "2026-02-25T06:15:24.1995008Z",

"endTime": "2026-02-25T06:15:24.1995008Z",

"triggerName": "manual",

"triggerName": "manual",

"error": {"code": "ActionFailed", "message": "An action failed..."}},

"error": {"code": "ActionFailed", "message": "An action failed..."}},

{"name": "08584296028664130474944675379CU26",

{"name": "08584296028664130474944675379CU26",

"status": "Succeeded", "error": null, ...}]

"status": "Succeeded", "error": null, ...}]

for r in runs: print(r["name"], r["status"])
for r in runs: print(r["name"], r["status"])

Get the name of the first failed run

获取第一个失败运行的名称

run_id = next((r["name"] for r in runs if r["status"] == "Failed"), None)

---
run_id = next((r["name"] for r in runs if r["status"] == "Failed"), None)

---

Inspect an Action's Output

查看操作输出

python
run_id = runs[0]["name"]

out = mcp("get_live_flow_run_action_outputs", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id,
    "actionName": "Get_Customer_Record"   # exact action name from the definition
})
print(json.dumps(out, indent=2))

python
run_id = runs[0]["name"]

out = mcp("get_live_flow_run_action_outputs", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id,
    "actionName": "Get_Customer_Record"   # 流定义中的精确操作名称
})
print(json.dumps(out, indent=2))

Get a Run's Error

获取运行错误信息

python
err = mcp("get_live_flow_run_error", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})
python
err = mcp("get_live_flow_run_error", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})

Returns:

返回:

{"runName": "08584296068...",

{"runName": "08584296068...",

"failedActions": [

"failedActions": [

{"actionName": "HTTP_find_AD_User_by_Name", "status": "Failed",

{"actionName": "HTTP_find_AD_User_by_Name", "status": "Failed",

"code": "NotSpecified", "startTime": "...", "endTime": "..."},

"code": "NotSpecified", "startTime": "...", "endTime": "..."},

{"actionName": "Scope_prepare_workers", "status": "Failed",

{"actionName": "Scope_prepare_workers", "status": "Failed",

"error": {"code": "ActionFailed", "message": "An action failed..."}}

"error": {"code": "ActionFailed", "message": "An action failed..."}}

],

],

"allActions": [

"allActions": [

{"actionName": "Apply_to_each", "status": "Skipped"},

{"actionName": "Apply_to_each", "status": "Skipped"},

{"actionName": "Compose_WeekEnd", "status": "Succeeded"},

{"actionName": "Compose_WeekEnd", "status": "Succeeded"},

...

...

]}

]}

The ROOT cause is usually the deepest entry in failedActions:

根本原因通常是failedActions中的最后一条记录:

root = err["failedActions"][-1] print(f"Root failure: {root['actionName']} → {root['code']}")

---
root = err["failedActions"][-1] print(f"Root failure: {root['actionName']} → {root['code']}")

---

Resubmit a Run

重新提交运行任务

python
result = mcp("resubmit_live_flow_run", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})
print(result)   # {"resubmitted": true, "triggerName": "..."}

python
result = mcp("resubmit_live_flow_run", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})
print(result)   # {"resubmitted": true, "triggerName": "..."}

Cancel a Running Run

取消正在运行的任务

python
mcp("cancel_live_flow_run", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})
⚠️ Do NOT cancel a run that shows
Running
because it is waiting for an adaptive card response.
That status is normal — the flow is paused waiting for a human to respond in Teams. Cancelling it will discard the pending card.

python
mcp("cancel_live_flow_run", {
    "environmentName": ENV,
    "flowName": FLOW,
    "runName": run_id
})
⚠️ 如果运行状态显示
Running
是因为正在等待自适应卡片响应,请勿取消。
该状态是正常的:流已暂停,等待用户在Teams中响应。取消会丢弃待处理的卡片。

Full Round-Trip Example — Debug and Fix a Failing Flow

完整端到端示例 — 调试并修复失败的流

python
undefined
python
undefined

── 1. Find the flow ─────────────────────────────────────────────────────

── 1. 查找目标流 ─────────────────────────────────────────────────────

result = mcp("list_live_flows", {"environmentName": ENV}) target = next(f for f in result["flows"] if "My Flow Name" in f["displayName"]) FLOW_ID = target["id"]
result = mcp("list_live_flows", {"environmentName": ENV}) target = next(f for f in result["flows"] if "My Flow Name" in f["displayName"]) FLOW_ID = target["id"]

── 2. Get the most recent failed run ────────────────────────────────────

── 2. 获取最近的失败运行记录 ────────────────────────────────────

runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW_ID, "top": 5})
runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW_ID, "top": 5})

[{"name": "08584296068...", "status": "Failed", ...}, ...]

[{"name": "08584296068...", "status": "Failed", ...}, ...]

RUN_ID = next(r["name"] for r in runs if r["status"] == "Failed")
RUN_ID = next(r["name"] for r in runs if r["status"] == "Failed")

── 3. Get per-action failure breakdown ──────────────────────────────────

── 3. 获取按操作拆分的失败详情 ──────────────────────────────────

err = mcp("get_live_flow_run_error", {"environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID})
err = mcp("get_live_flow_run_error", {"environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID})

{"failedActions": [{"actionName": "HTTP_find_AD_User_by_Name", "code": "NotSpecified",...}], ...}

{"failedActions": [{"actionName": "HTTP_find_AD_User_by_Name", "code": "NotSpecified",...}], ...}

root_action = err["failedActions"][-1]["actionName"] print(f"Root failure: {root_action}")
root_action = err["failedActions"][-1]["actionName"] print(f"Root failure: {root_action}")

── 4. Read the definition and inspect the failing action's expression ───

── 4. 读取流定义,查看失败操作的表达式 ───

defn = mcp("get_live_flow", {"environmentName": ENV, "flowName": FLOW_ID}) acts = defn["properties"]["definition"]["actions"] print("Failing action inputs:", acts[root_action]["inputs"])
defn = mcp("get_live_flow", {"environmentName": ENV, "flowName": FLOW_ID}) acts = defn["properties"]["definition"]["actions"] print("Failing action inputs:", acts[root_action]["inputs"])

── 5. Inspect the prior action's output to find the null ────────────────

── 5. 查看前一个操作的输出,找到空值 ────────────────

out = mcp("get_live_flow_run_action_outputs", { "environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID, "actionName": "Compose_Names" }) nulls = [x for x in out.get("body", []) if x.get("Name") is None] print(f"{len(nulls)} records with null Name")
out = mcp("get_live_flow_run_action_outputs", { "environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID, "actionName": "Compose_Names" }) nulls = [x for x in out.get("body", []) if x.get("Name") is None] print(f"{len(nulls)} records with null Name")

── 6. Apply the fix ─────────────────────────────────────────────────────

── 6. 应用修复 ─────────────────────────────────────────────────────

acts[root_action]["inputs"]["parameters"]["searchName"] =
"@coalesce(item()?['Name'], '')"
conn_refs = defn["properties"]["connectionReferences"] result = mcp("update_live_flow", { "environmentName": ENV, "flowName": FLOW_ID, "definition": defn["properties"]["definition"], "connectionReferences": conn_refs }) assert result.get("error") is None, f"Deploy failed: {result['error']}"
acts[root_action]["inputs"]["parameters"]["searchName"] =
"@coalesce(item()?['Name'], '')"
conn_refs = defn["properties"]["connectionReferences"] result = mcp("update_live_flow", { "environmentName": ENV, "flowName": FLOW_ID, "definition": defn["properties"]["definition"], "connectionReferences": conn_refs }) assert result.get("error") is None, f"Deploy failed: {result['error']}"

⚠️ error key is always present — only fail if it is NOT None

⚠️ error字段始终存在,仅当它不为None时才代表部署失败

── 7. Resubmit and verify ───────────────────────────────────────────────

── 7. 重新提交并验证结果 ───────────────────────────────────────────────

mcp("resubmit_live_flow_run", {"environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID})
import time; time.sleep(30) new_runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW_ID, "top": 1}) print(new_runs[0]["status"]) # Succeeded = done

---
mcp("resubmit_live_flow_run", {"environmentName": ENV, "flowName": FLOW_ID, "runName": RUN_ID})
import time; time.sleep(30) new_runs = mcp("get_live_flow_runs", {"environmentName": ENV, "flowName": FLOW_ID, "top": 1}) print(new_runs[0]["status"]) # 显示Succeeded即完成

---

Auth & Connection Notes

鉴权与连接说明

FieldValue
Auth header
x-api-key: <JWT>
not
Authorization: Bearer
Token formatPlain JWT — do not strip, alter, or prefix it
TimeoutUse ≥ 120 s for
get_live_flow_run_action_outputs
(large outputs)
Environment name
Default-<tenant-guid>
(find it via
list_live_environments
or
list_live_flows
response)

字段
鉴权请求头
x-api-key: <JWT>
不是
Authorization: Bearer
令牌格式纯JWT,请勿裁剪、修改或添加前缀
超时设置
get_live_flow_run_action_outputs
请设置≥120秒的超时(输出可能很大)
环境名称
Default-<租户GUID>
(可通过
list_live_environments
list_live_flows
响应获取)

Reference Files

参考文件

  • MCP-BOOTSTRAP.md — endpoint, auth, request/response format (read this first)
  • tool-reference.md — response shapes and behavioral notes (parameters are in
    tools/list
    )
  • action-types.md — Power Automate action type patterns
  • connection-references.md — connector reference guide

  • MCP-BOOTSTRAP.md — 端点、鉴权、请求/响应格式(请先阅读)
  • tool-reference.md — 响应结构和行为说明(参数定义见
    tools/list
  • action-types.md — Power Automate操作类型模式
  • connection-references.md — 连接器参考指南

More Capabilities

更多功能

For diagnosing failing flows end-to-end → load the
power-automate-debug
skill.
For building and deploying new flows → load the
power-automate-build
skill.
如需端到端诊断失败的流 → 加载
power-automate-debug
技能。
如需构建和部署新流 → 加载
power-automate-build
技能。