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(fMCP HTTP {e.code}: {body[:200]}) from e
raw = json.loads(resp.read())
if error in raw:
raise RuntimeError(fMCP error: {json.dumps(raw[error])})
return json.loads(raw[result][content][0][text])

ENV = # 例如 Default-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx

FlowStudio Teams版：快速诊断路径（跳过步骤2-4）

如果您拥有FlowStudio Teams版订阅，getstoreflow_errors可在单次调用中返回每次运行的失败数据，包括操作名称和修复提示——无需逐步遍历实时API步骤。

python

快速失败摘要

{totalRuns: 100, failRuns: 10, failRate: 0.1,

averageDurationSeconds: 29.4, maxDurationSeconds: 158.9,

firstFailRunRemediation: }

每次运行的错误详情（需要配置活动监控）

获取完整的治理记录（描述、复杂度、层级、连接器列表）：
python
record = mcp(getstoreflow, environmentName=ENV, flowName=FLOW_ID)

{displayName: My Flow, state: Started,

runPeriodTotal: 100, runPeriodFailRate: 0.1, runPeriodFails: 10,

runPeriodDurationAverage: 29410.8, ← 毫秒

runError: {\code\: \EACCES\, ...}, ← JSON字符串，需解析

description: ..., tier: Premium, complexity: {...}}

步骤1 — 定位流

python
result = mcp(listliveflows, environmentName=ENV)

返回一个包装对象：{mode, flows, totalCount, error}

步骤2 — 查找失败运行

python
runs = mcp(getliveflowruns, environmentName=ENV, flowName=FLOWID, top=5)

返回直接数组（最新在前）：

[{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 — 获取顶层错误

python
err = mcp(getliveflowrunerror,
environmentName=ENV, flowName=FLOWID, runName=RUNID)

返回：

{

runName: 08584296068667933411438594643CU15,

failedActions: [

{actionName: Applytoeachprepareworkers, status: Failed,

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

startTime: ..., endTime: ...},

{actionName: HTTPfindADUserby_Name, status: Failed,

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

],

allActions: [

{actionName: Applytoeach, status: Skipped},

{actionName: Compose_WeekEnd, status: Succeeded},

...

]

}

failedActions按从外到内排序。根本原因是最后一个条目：

allActions显示每个操作的状态 — 有助于发现哪些被跳过

参见common-errors.md以解码错误代码。

步骤4 — 读取流定义

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

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

步骤5 — 检查操作输出（从失败回溯）

对于导致失败的每个操作，检查其运行时输出：

python
for actionname in [ComposeWeekEnd, HTTPGetData, Parse_JSON]:
result = mcp(getliveflowrunaction_outputs,
environmentName=ENV,
flowName=FLOW_ID,
runName=RUN_ID,
actionName=action_name)
# 返回数组 — 提供actionName时为单元素
out = result[0] if result else {}
print(action_name, out.get(status))
print(json.dumps(out.get(outputs, {}), indent=2)[:500])

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

步骤6 — 定位根本原因

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

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

python

示例：操作使用 split(item()?[Name], )

→ 源数据中Name为null

返回单元素数组；索引[0]获取操作对象