mcpField
P
power-automate-debug

power-automate-debug

github.com/ninihen1开发3440 次安装8 周前更新
代码仓库

关于 power-automate-debug

这是一个通过 FlowStudio MCP 排查 Power Automate 云端流程故障的诊断技能,供 Agent 定位失败运行、检查动作数据并分析根因,也涵盖修复与验证步骤。

主要功能

  1. 01定位流程和失败运行
  2. 02检查动作输入与输出
  3. 03追踪上游数据和表达式
  4. 04应用并验证流程修复

适用场景

  1. 01诊断空值或字段错误
  2. 02查找 HTTP 动作失败原因
  3. 03重放失败运行以测试修复

MCP 相关

需要可访问且具有有效 JWT 的 FlowStudio MCP 服务。该技能调用服务的流程和运行工具来检查故障、更新流程并重新提交运行。

安装

复制下面的命令,在终端中运行。

$npx skills add https://github.com/ninihen1/power-automate-mcp-skills --skill power-automate-debug

安装后在 Claude Code 中输入/power-automate-debug确认已加载。

附带文件references/common-errors.mdreferences/debug-workflow.md

详细介绍

使用 FlowStudio MCP 调试 Power Automate

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

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

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


事实来源

务必先调用 list_skills / tool_search,以确认可用的工具名称 和参数架构。工具名称和参数可能会随服务版本变化。 本技能介绍响应格式、行为说明和诊断模式——这些内容无法从工具架构中获知。 如果本文档与 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 密钥。


快速参考:诊断决策树

症状首选工具然后务必调用查看内容
流显示为失败get_live_flow_run_error对失败的操作调用 get_live_flow_run_action_outputsoutputs 中的 HTTP 状态码和响应正文
错误代码过于笼统(ActionFailed、NotSpecified)—get_live_flow_run_action_outputsoutputs.body 中包含实际错误消息、堆栈跟踪或 API 错误
HTTP 操作返回 500—get_live_flow_run_action_outputsoutputs.statusCode 和 outputs.body 中的服务器错误详情
表达式崩溃—对前一个操作调用 get_live_flow_run_action_outputs输出正文中的 null / 类型错误字段
流无法启动get_live_flow—检查 properties.state 是否为 "Started"
操作返回错误数据get_live_flow_run_action_outputs—实际输出正文与预期是否相符
已应用修复但仍然失败重新提交后调用 get_live_flow_runs—新运行的 status 字段

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


参考文件

相关技能

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

评论

支持 Markdown