flowstudio-power-automate-debug

使用 FlowStudio MCP 调试 Power Automate

通过 FlowStudio MCP 服务排查失败的 Power Automate 云端流的逐步诊断流程。

真实调试示例：子流中的表达式错误 | 数据录入问题，而非流缺陷 | 空值导致子流崩溃

前提条件：必须能够使用有效的 JWT 访问 FlowStudio MCP 服务。 连接设置请参阅 flowstudio-power-automate-mcp 技能。 在 https://mcp.flowstudio.app 订阅

---

事实来源

务必先调用 list_skills / tool_search，以确认可用的工具名称和参数 schema。工具名称和参数可能因服务版本而异。 本技能涵盖响应结构、行为说明和诊断模式——这些是工具 schema 无法告知你的内容。如果本文档与 tool_search 或实际 API 响应不一致，以 API 为准。

---

Python 辅助函数

import json, urllib.request

MCP_URL   = "https://mcp.flowstudio.app/mcp"
MCP_TOKEN = "<YOUR_JWT_TOKEN>"

def mcp(tool, **kwargs):
    payload = json.dumps({"jsonrpc": "2.0", "id": 1, "method": "tools/call",
                          "params": {"name": tool, "arguments": kwargs}}).encode()
    req = urllib.request.Request(MCP_URL, data=payload,
        headers={"x-api-key": MCP_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'])}")
    return json.loads(raw["result"]["content"][0]["text"])

ENV = "<environment-id>"   # e.g. Default-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

---

第 1 步 — 定位流

result = mcp("list_live_flows", environmentName=ENV)
# Returns a wrapper object: {mode, flows, totalCount, error}
target = next(f for f in result["flows"] if "My Flow Name" in f["displayName"])
FLOW_ID = target["id"]   # plain UUID — use directly as flowName
print(FLOW_ID)

---

第 2 步 — 查找失败的运行

runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=5)
# Returns direct array (newest first):
# [{"name": "08584296068667933411438594643CU15",
#   "status": "Failed",
#   "startTime": "2026-02-25T06:13:38.6910688Z",
#   "endTime": "2026-02-25T06:15:24.1995008Z",
#   "triggerName": "manual",
#   "error": {"code": "ActionFailed", "message": "An action failed..."}},
#  {"name": "...", "status": "Succeeded", "error": null, ...}]

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

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

---

第 3 步 — 获取顶层错误

关键：get_live_flow_run_error 告诉你哪个操作失败了。 get_live_flow_run_action_outputs 告诉你为什么失败。你必须同时调用两者。 绝不要仅停留在错误本身——像 ActionFailed、 NotSpecified 和 InternalServerError 这样的错误代码只是通用包装。实际的 根本原因（字段错误、空值、HTTP 500 响应体、堆栈跟踪）只有在 操作的输入和输出中才可见。

err = mcp("get_live_flow_run_error",
    environmentName=ENV, flowName=FLOW_ID, runName=RUN_ID)
# Returns:
# {
#   "runName": "08584296068667933411438594643CU15",
#   "failedActions": [
#     {"actionName": "Apply_to_each_prepare_workers", "status": "Failed",
#      "error": {"code": "ActionFailed", "message": "An action failed..."},
#      "startTime": "...", "endTime": "..."},
#     {"actionName": "HTTP_find_AD_User_by_Name", "status": "Failed",
#      "code": "NotSpecified", "startTime": "...", "endTime": "..."}
#   ],
#   "allActions": [
#     {"actionName": "Apply_to_each", "status": "Skipped"},
#     {"actionName": "Compose_WeekEnd", "status": "Succeeded"},
#     ...
#   ]
# }

# failedActions is ordered outer-to-inner. The ROOT cause is the LAST entry:
root = err["failedActions"][-1]
print(f"Root action: {root['actionName']} → code: {root.get('code')}")

# allActions shows every action's status — useful for spotting what was Skipped
# See common-errors.md to decode the error code.

---

第 4 步 — 检查失败操作的输入和输出

这是最重要的一步。 get_live_flow_run_error 只给你 一个通用错误代码。实际的错误详情——HTTP 状态码、 响应体、堆栈跟踪、空值——存在于操作的运行时 输入和输出中。在识别出失败操作后，务必立即检查它。

# Get the root failing action's full inputs and outputs
root_action = err["failedActions"][-1]["actionName"]
detail = mcp("get_live_flow_run_action_outputs",
    environmentName=ENV,
    flowName=FLOW_ID,
    runName=RUN_ID,
    actionName=root_action)

if len(detail) > 1:
    print(f"{root_action} returned {len(detail)} repetitions; inspect iteration indexes")
out = detail[0] if detail else {}
print(f"Action: {out.get('actionName')}")
print(f"Status: {out.get('status')}")

# For HTTP actions, the real error is in outputs.body
if isinstance(out.get("outputs"), dict):
    status_code = out["outputs"].get("statusCode")
    body = out["outputs"].get("body", {})
    print(f"HTTP {status_code}")
    print(json.dumps(body, indent=2)[:500])

    # Error bodies are often nested JSON strings — parse them
    if isinstance(body, dict) and "error" in body:
        err_detail = body["error"]
        if isinstance(err_detail, str):
            err_detail = json.loads(err_detail)
        print(f"Error: {err_detail.get('message', err_detail)}")

# For expression errors, the error is in the error field
if out.get("error"):
    print(f"Error: {out['error']}")

# Also check inputs — they show what expression/URL/body was used
if out.get("inputs"):
    print(f"Inputs: {json.dumps(out['inputs'], indent=2)[:500]}")

操作输出揭示的内容（错误代码无法揭示的）

来自 get_live_flow_run_error 的错误代码	get_live_flow_run_action_outputs 揭示的内容
ActionFailed	实际失败的嵌套操作及其 HTTP 响应
NotSpecified	HTTP 状态码 + 包含真实错误的响应体
InternalServerError	服务器的错误消息、堆栈跟踪或 API 错误 JSON
InvalidTemplate	失败的确切表达式以及空值/类型错误的值
BadRequest	发送的请求体以及服务器拒绝它的原因

Foreach 迭代

当 actionName 指向 foreach 内部的某个操作时，输出工具可以 返回该操作的每一次重复。每个项可能包含 repetitionIndexes，其中含有循环名称和从零开始的 itemIndex。在找到可疑项后， 使用 iterationIndex 检查某一次迭代：

all_reps = mcp("get_live_flow_run_action_outputs",
    environmentName=ENV,
    flowName=FLOW_ID,
    runName=RUN_ID,
    actionName=root_action)

for rep in all_reps[:10]:
    print(rep.get("repetitionIndexes"), rep.get("status"), rep.get("error"))

one_rep = mcp("get_live_flow_run_action_outputs",
    environmentName=ENV,
    flowName=FLOW_ID,
    runName=RUN_ID,
    actionName=root_action,
    iterationIndex=3)

证据 Compose 书挡

对于不确定的连接器工作，在风险操作之前添加一个 Compose_*_Request， 在其之后添加一个 Compose_*_Result，并允许结果操作在 Succeeded 和 Failed 两种情况下执行。这样可以为将来的调试提供干净的载荷快照， 而无需再次部署。不要在这些书挡中包含机密或较长的二进制载荷。

示例：HTTP 操作返回 500

Error code: "InternalServerError" ← this tells you nothing

Action outputs reveal:
  HTTP 500
  body: {"error": "Cannot read properties of undefined (reading 'toLowerCase')
    at getClientParamsFromConnectionString (storage.js:20)"}
  ← THIS tells you the Azure Function crashed because a connection string is undefined

示例：空值导致的表达式错误

Error code: "BadRequest" ← generic

Action outputs reveal:
  inputs: "body('HTTP_GetTokenFromStore')?['token']?['access_token']"
  outputs: ""   ← empty string, the path resolved to null
  ← THIS tells you the response shape changed — token is at body.access_token, not body.token.access_token

---

第 5 步 — 阅读流定义

defn = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
actions = defn["properties"]["definition"]["actions"]
print(list(actions.keys()))

在定义中找到失败的操作。检查它的 inputs 表达式， 以了解它期望什么数据。

---

步骤 6 — 从失败处回溯

当失败操作的输入引用了上游操作时，也要检查这些上游操作。沿着链路 向后回溯，直到找到坏数据的来源：

# Inspect multiple actions leading up to the failure
for action_name in [root_action, "Compose_WeekEnd", "HTTP_Get_Data"]:
    result = mcp("get_live_flow_run_action_outputs",
        environmentName=ENV,
        flowName=FLOW_ID,
        runName=RUN_ID,
        actionName=action_name)
    out = result[0] if result else {}
    print(f"\n--- {action_name} ({out.get('status')}) ---")
    print(f"Inputs:  {json.dumps(out.get('inputs', ''), indent=2)[:300]}")
    print(f"Outputs: {json.dumps(out.get('outputs', ''), indent=2)[:300]}")

⚠️ 数组处理操作的输出负载可能非常大。 打印前务必先切片（例如 [:500]）。

提示：当你不确定是哪个操作产生了坏数据时，省略 actionName 以列出顶层操作。一旦你选中了 foreach 内的某个操作， 传入 iterationIndex 以避免把每次重复都拉入上下文。

---

步骤 7 — 定位根本原因

表达式错误（例如对 null 执行 split）

如果错误提到 InvalidTemplate 或某个函数名：

1. 在定义中找到该操作
2. 检查它读取了哪个上游操作/表达式
3. 检查该上游操作的输出，看是否有 null / 缺失字段

# Example: action uses split(item()?['Name'], ' ')
# → null Name in the source data
result = mcp("get_live_flow_run_action_outputs", ..., actionName="Compose_Names")
if not result:
    print("No outputs returned for Compose_Names")
    names = []
else:
    names = result[0].get("outputs", {}).get("body") or []
nulls = [x for x in names if x.get("Name") is None]
print(f"{len(nulls)} records with null Name")

字段路径错误

表达式 triggerBody()?['fieldName'] 返回 null → fieldName 错了。 检查触发器输出以查看实际字段名：

result = mcp("get_live_flow_run_action_outputs", ..., actionName="<trigger-action-name>")
print(json.dumps(result[0].get("outputs"), indent=2)[:500])

HTTP 操作返回错误

错误码显示 InternalServerError 或 NotSpecified — 务必检查 操作输出以获取实际的 HTTP 状态和响应正文：

result = mcp("get_live_flow_run_action_outputs", ..., actionName="HTTP_Get_Data")
out = result[0]
print(f"HTTP {out['outputs']['statusCode']}")
print(json.dumps(out['outputs']['body'], indent=2)[:500])

连接 / 认证失败

查找 ConnectionAuthorizationFailed — 连接所有者必须与运行该流的 服务账户一致。无法通过 API 修复；请在 PA 设计器中修复。

Outlook 用户选择器失败（DynamicListValuesUndefinedOrInvalid）

GetEmailsV3 等 Outlook 操作使用的参数（mailboxAddress、to、cc、 from）其下拉列表由 builtInOperation:AadGraph.GetUsers 支撑 — 该操作 在 PA listEnum 层已损坏，始终返回 DynamicListValuesUndefinedOrInvalid。当 Agent 通过 update_live_flow 重建或修改 Outlook 操作，并尝试通过动态选项解析用户时，就会出现此问题。 不要通过重试 AadGraph 来修复 — 改用 shared_office365users.SearchUserV2（返回相同的 AAD 用户结构）。 使用 describe_live_connector 确认受影响的参数是否暴露了结构化的 fallback，然后对 shared_office365users.SearchUserV2 调用 get_live_dynamic_options，而不是对已损坏的 AadGraph 操作调用。 对于动态字段架构而非下拉选项，使用 get_live_dynamic_properties， 并传入 describe_live_connector 返回的元数据。

---

步骤 8 — 应用修复

针对表达式/数据问题：

defn = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
acts = defn["properties"]["definition"]["actions"]

# Example: fix split on potentially-null Name
acts["Compose_Names"]["inputs"] = \
    "@coalesce(item()?['Name'], 'Unknown')"

conn_refs = defn["properties"]["connectionReferences"]
result = mcp("update_live_flow",
    environmentName=ENV,
    flowName=FLOW_ID,
    definition=defn["properties"]["definition"],
    connectionReferences=conn_refs)

print(result.get("error"))  # None = success

⚠️ update_live_flow 总是返回一个 error 键。 值为 null（Python 中的 None）表示成功。

---

步骤 9 — 验证修复

使用 resubmit_live_flow_run 测试任何流 — 不只是 HTTP 触发器。 resubmit_live_flow_run 使用原始触发器负载重放之前的运行。 这适用于所有触发器类型：Recurrence、SharePoint "创建项时"、连接器 webhook、Button 触发器以及 HTTP 触发器。你无需让用户手动触发该流，也无需 等待下一次计划运行。

唯一无法使用 resubmit 的情况是从未运行过的全新流 — 它没有可重放的先前运行。

# Resubmit the failed run — works for ANY trigger type
resubmit = mcp("resubmit_live_flow_run",
    environmentName=ENV, flowName=FLOW_ID, runName=RUN_ID)
print(resubmit)   # {"resubmitted": true, "triggerName": "..."}

# Wait ~30 s then check
import time; time.sleep(30)
new_runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=3)
print(new_runs[0]["status"])   # Succeeded = done

何时使用 resubmit 与 trigger

场景	使用	原因
在任意流上测试修复	resubmit_live_flow_run	重放导致失败的确切触发器负载 — 最佳验证方式
Recurrence / 计划流	trigger_live_flow（无 body）	立即运行，类似门户的"运行流"按钮；resubmit 重放过去运行的数据
SharePoint / 连接器触发器	resubmit_live_flow_run	不创建真实的 SP 项就无法触发
HTTP、Button 或 PowerApps 触发器，带自定义测试负载	trigger_live_flow	当你需要发送与原始运行不同的数据时
从未运行过的全新流	trigger_live_flow	没有可重放的先前运行

使用自定义负载测试 HTTP、Button 和 PowerApps 流

对于带有 Request 触发器的流（HTTP 请求、手动 Button 或 PowerApps）， 当你需要发送与原始运行不同的负载时，使用 trigger_live_flow。 对于每种类型，都将触发器输入作为 body 传入：

# First inspect what the trigger expects — read directly from the flow definition
defn = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
triggers = defn["properties"]["definition"]["triggers"]
manual = next(iter(triggers.values()))   # usually the only trigger on HTTP flows
request_schema = manual.get("inputs", {}).get("schema")
print("Expected body schema:", request_schema)

# Response schemas live on Response action(s) in the actions block
for name, act in defn["properties"]["definition"]["actions"].items():
    if act.get("type") == "Response":
        print(f"Response {name}:", act.get("inputs", {}).get("schema"))

# Trigger with a test payload
result = mcp("trigger_live_flow",
    environmentName=ENV,
    flowName=FLOW_ID,
    body={"name": "Test User", "value": 42})
print(f"Status: {result['responseStatus']}, Body: {result.get('responseBody')}")
print(f"Kind: {result['triggerKind']}, via: {result['invocation']}, run: {result.get('runName')}")
if result.get("warning"):
    print(result["warning"])   # required trigger inputs you left out

trigger_live_flow 会自动处理 AAD 身份验证的触发器。 适用于 Request 触发器（HTTP 请求、Button、PowerApps）以及 计划（Recurrence）流，后者会立即运行——不需要 body， 因为计划触发器不接受输入（提供 body 会被拒绝）。自动化 连接器触发器只会从其源事件触发。

Power Automate 不会强制执行触发器的 required 输入。如果你遗漏了 某个输入，运行仍会启动，该输入为 null，结果中会带有 一个 warning，指出缺失的键。如果这很重要，请取消该次运行并使用 完整的 body 再次调用。

runName 仅在 Button 和 PowerApps 运行中返回。对于 HTTP 触发器， 请使用 get_live_flow_runs 查找运行。

通过浏览器扩展密钥时，Button 和 PowerApps 触发器只能以 空 body 运行。工具会说明这一点，并列出解决办法：重新提交过去的运行、 在流内使用 coalesce(triggerBody()?['x'], 'value') 为输入设置默认值， 或使用标准 API 密钥。

---

快速参考诊断决策树

症状	首选工具	然后始终调用	需要查看什么
流显示为 Failed	get_live_flow_run_error	对失败的操作调用 get_live_flow_run_action_outputs	outputs 中的 HTTP 状态 + 响应 body
错误代码是通用代码（ActionFailed、NotSpecified）	—	get_live_flow_run_action_outputs	outputs.body 包含真正的错误消息、堆栈跟踪或 API 错误
HTTP 操作返回 500	—	get_live_flow_run_action_outputs	outputs.statusCode + 包含服务器错误详情的 outputs.body
表达式崩溃	—	对先前的操作调用 get_live_flow_run_action_outputs	输出 body 中的 null / 类型错误的字段
流从未启动	get_live_flow	—	检查 properties.state = "Started"
操作返回错误数据	get_live_flow_run_action_outputs	—	实际输出 body 与预期对比
已应用修复但仍失败	重新提交后调用 get_live_flow_runs	—	新运行的 status 字段

规则：绝不要仅凭错误代码进行诊断。 get_live_flow_run_error 会指出失败的操作。get_live_flow_run_action_outputs 会揭示 实际原因。务必同时调用两者。

---

参考文件

• common-errors.md — 错误代码、可能的原因和修复方法
• debug-workflow.md — 用于复杂故障的完整决策树

相关技能

• flowstudio-power-automate-mcp — 基础技能：连接设置、MCP 辅助工具、工具发现
• flowstudio-power-automate-build — 构建和部署新流