power-automate-build

使用 FlowStudio MCP 构建和部署 Power Automate 流

通过 FlowStudio MCP 服务，以编程方式构建和部署 Power Automate 云端流的分步指南。

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

工作流：

1. 加载当前构建工具。
2. 检查是否已有流。
3. 解析连接引用。
4. 构建定义。
5. 部署。
6. 验证。
7. 测试。

---

权威来源

务必先调用 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

---

0. 加载当前构建工具

对于全新流，加载服务的 create-flow 工具包。对于编辑现有流，加载 build-flow。这样可以确保 Agent 在构建 JSON 之前，与 MCP 服务的当前架构保持一致。

schemas = mcp("tool_search", query="skill:create-flow")
# Includes list_live_environments, list_live_connections,
# describe_live_connector, get_live_dynamic_options, update_live_flow.

如果需要工具包以外的工具，请显式加载：

mcp("tool_search", query="select:get_live_dynamic_properties")

---

1. 安全检查：流是否已存在？

构建前务必先检查，以避免重复：

results = mcp("list_live_flows",
    environmentName=ENV,
    mode="owner",
    search="My New Flow",
    top=20)

# list_live_flows returns { "flows": [...], "mode": "...", ... }
matches = [f for f in results["flows"]
           if "My New Flow".lower() in f["displayName"].lower()]

if len(matches) > 0:
    # Flow exists — modify rather than create
    FLOW_ID = matches[0]["id"]   # plain UUID from list_live_flows
    print(f"Existing flow: {FLOW_ID}")
    defn = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
else:
    print("Flow not found — building from scratch")
    FLOW_ID = None

对于非常庞大的环境，list_live_flows 可能会返回续传 URL。将其作为 continuationUrl，并使用相同的 mode 传回，以获取下一批结果。仅当用户需要查看环境中的所有流，且 MCP 身份具有管理员权限时，才使用 mode="admin"。

---

2. 获取连接引用

每个连接器操作都需要一个 connectionName，该名称指向流的 connectionReferences 映射中的一个键。该键链接到环境中的一个已通过身份验证的连接。

必需：你必须先调用 list_live_connections——不要向用户询问连接名称或 GUID。API 会返回所需的确切值。 只有在 API 确认缺少必需连接时，才提示用户。

2a — 查找活动连接

conns = mcp("list_live_connections", environmentName=ENV)
active = [c for c in conns["connections"]
          if c["statuses"][0]["status"] == "Connected"]
conn_map = {c["connectorName"]: c["id"] for c in active}

对于已知连接器，传入 search 以缩减输出，并获取可直接粘贴的 connectionReferenceTemplate 和 hostTemplate 值：

sp_conns = mcp("list_live_connections",
    environmentName=ENV,
    search="shared_sharepointonline")

2b — 确定流需要哪些连接器

常见的连接器 API 名称：SharePoint shared_sharepointonline、Outlook shared_office365、Teams shared_teams、Approvals shared_approvals、 OneDrive shared_onedriveforbusiness、Excel shared_excelonlinebusiness、 Dataverse shared_commondataserviceforapps、Forms shared_microsoftforms。

不需要连接器的流（例如仅使用 Recurrence + Compose + HTTP 的流）可以省略 connectionReferences。

2c — 如果缺少连接，请引导用户

connectors_needed = ["shared_sharepointonline", "shared_office365"]  # adjust per flow
missing = [c for c in connectors_needed if c not in conn_map]
if missing:
    # STOP: connections require browser OAuth consent.
    # Ask the user to create the missing connector connections in the
    # selected environment, then re-run list_live_connections.
    raise Exception(f"Missing active connections: {missing}")

2d — 构建 connectionReferences 块

connection_references = {}
host_templates = {}
for connector in connectors_needed:
    c = next(c for c in active if c["connectorName"] == connector)
    connection_references[connector] = c.get("connectionReferenceTemplate") or {
        "connectionName": c["id"],   # the connection id from list_live_connections
        "source": "Invoker",
        "id": f"/providers/Microsoft.PowerApps/apis/{connector}"
    }
    host_templates[connector] = c.get("hostTemplate") or {
        "connectionName": connector
    }

在第 3 步的操作 JSON 中，inputs.host.connectionName 必须是映射键，例如 shared_teams，而不是 GUID。GUID 只应出现在 connectionReferences[connector].connectionName 值中。如果现有流使用 相同的连接器，也可以从 get_live_flow 复制其 properties.connectionReferences。

---

3. 构建流定义

构建定义对象。完整架构请参阅 flow-schema.md，并参阅以下操作模式文档，获取可直接复制粘贴的模板：

• action-patterns-core.md — 变量、控制流、表达式
• action-patterns-data.md — 数组转换、HTTP、解析
• action-patterns-connectors.md — SharePoint、Outlook、Teams、Approvals

definition = {
    "$schema": "https://schema.management.azure.com/providers/Microsoft.Logic/schemas/2016-06-01/workflowdefinition.json#",
    "contentVersion": "1.0.0.0",
    "triggers": { ... },   # see trigger-types.md / build-patterns.md
    "actions": { ... }     # see ACTION-PATTERNS-*.md / build-patterns.md
}

请参阅 build-patterns.md，其中提供了完整且可直接使用的流定义，涵盖 Recurrence+SharePoint+Teams、HTTP 触发器等。

在猜测 JSON 之前，先发现连接器操作

对于由连接器支持的触发器和操作，优先使用实时连接器描述器，而不是 手写结构。它可以返回作者提供的提示、规范示例、变体 键、输入/输出以及动态元数据指针。

# Search across connectors when you know the user's intent but not the API.
matches = mcp("describe_live_connector",
    environmentName=ENV,
    search="send email",
    top=5)

# Describe a specific operation before copying an exampleDefinition.
op = mcp("describe_live_connector",
    environmentName=ENV,
    connectorName="shared_office365",
    operationId="SendEmailV2")
print(op.get("hint"))

当一个操作有多个作者提供的变体时，请求流程 所需的那个变体：

teams_chat = mcp("describe_live_connector",
    environmentName=ENV,
    connectorName="shared_teams",
    operationId="PostMessageToConversation",
    variant="flowbot_chat")

当操作说明指出某个参数具有动态选项或动态 属性时，调用所指示的下一工具：

sp_op = mcp("describe_live_connector",
    environmentName=ENV,
    connectorName="shared_sharepointonline",
    operationId="GetItems")

sites = mcp("get_live_dynamic_options",
    environmentName=ENV,
    connectorName="shared_sharepointonline",
    connectionName=conn_map["shared_sharepointonline"],
    operationId="GetItems",
    parameterName="dataset",
    dynamicMetadata=sp_op["dynamicParameters"]["dataset"])

fields = mcp("get_live_dynamic_properties",
    environmentName=ENV,
    connectorName="shared_sharepointonline",
    connectionName=conn_map["shared_sharepointonline"],
    operationId="GetItems",
    parameterName="item",
    parameters={"dataset": "<site-url>", "table": "<list-id>"},
    dynamicMetadata=sp_op["dynamicProperties"]["item"])

对 SharePoint 网站/列表和 Teams 团队/频道等下拉 ID 使用动态选项。对 SharePoint 列表项列等 架构/字段结构使用动态属性。

---

4. 部署（创建或更新）

update_live_flow 在单个工具中同时处理创建和更新。

创建新流程（没有现有流程）

省略 flowName —— 服务会生成一个新的 GUID 并通过 PUT 创建：

definition["description"] = "Weekly SharePoint → Teams notification flow, built by agent"

result = mcp("update_live_flow",
    environmentName=ENV,
    # flowName omitted → creates a new flow
    definition=definition,
    connectionReferences=connection_references,
    displayName="Overdue Invoice Notifications"
)

if result.get("error") is not None:
    print("Create failed:", result["error"])
else:
    # Capture the new flow ID for subsequent steps
    FLOW_ID = result["created"]
    print(f"✅ Flow created: {FLOW_ID}")

更新现有流程

提供 flowName 以进行 PATCH：

definition["description"] = (
    "Updated by agent on " + __import__('datetime').datetime.utcnow().isoformat()
)

result = mcp("update_live_flow",
    environmentName=ENV,
    flowName=FLOW_ID,
    definition=definition,
    connectionReferences=connection_references,
    displayName="My Updated Flow"
)

if result.get("error") is not None:
    print("Update failed:", result["error"])
else:
    print("Update succeeded:", result)

⚠️ update_live_flow 始终返回一个 error 键。 null（Python 中的 None）表示成功 —— 不要将存在该键视为失败。

⚠️ 流程描述位于 definition["description"]。当前服务 会附加 #flowstudio-mcp 以进行使用跟踪。除非 tool_search 在活动架构中显示存在顶层 description 参数，否则不要传入它。

常见部署错误

错误消息（包含）	原因	修复
missing from connectionReferences	某个操作的 host.connectionName 引用了 connectionReferences 映射中不存在的键	确保 host.connectionName 使用 connectionReferences 中的键（例如 shared_teams），而不是原始 GUID
ConnectionAuthorizationFailed / 403	该连接 GUID 属于其他用户或未获授权	重新运行步骤 2a，并使用当前 x-api-key 用户拥有的连接
InvalidTemplate / InvalidDefinition	定义 JSON 中存在语法错误	检查 runAfter 链、表达式语法和操作类型拼写
ConnectionNotConfigured	连接器操作存在，但连接 GUID 无效或已过期	重新检查 list_live_connections 以获取新的 GUID

---

5. 验证部署

check = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)

# Confirm state
print("State:", check["properties"]["state"])  # Should be "Started"
# If state is "Stopped", use set_live_flow_state — NOT update_live_flow
# mcp("set_live_flow_state", environmentName=ENV, flowName=FLOW_ID, state="Started")

# Confirm the action we added is there
acts = check["properties"]["definition"]["actions"]
print("Actions:", list(acts.keys()))

---

6. 测试流程

强制要求：在触发任何测试运行之前，先向用户请求确认。 运行流程会产生真实的副作用 —— 它可能会发送电子邮件、发布 Teams 消息、 写入 SharePoint、启动审批或调用外部 API。说明该 流程将做什么，并在调用 trigger_live_flow 或 resubmit_live_flow_run 之前等待明确批准。

已更新的流程（有先前运行）—— 任何触发器类型

优先使用 resubmit_live_flow_run。 它适用于每一种触发器类型 —— Recurrence、SharePoint、连接器 webhook、Button 和 HTTP。它会重放 原始触发器负载。不要要求用户手动触发 流程或等待下一次计划运行。

runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=1)
if runs:
    # Works for Recurrence, SharePoint, connector triggers — not just HTTP
    result = mcp("resubmit_live_flow_run",
        environmentName=ENV, flowName=FLOW_ID, runName=runs[0]["name"])
    print(result)   # {"resubmitted": true, "triggerName": "..."}

HTTP、Button 和 PowerApps 流程 —— 自定义测试负载

仅当你需要发送与原始运行不同的负载时， 才使用 trigger_live_flow。对于验证修复，resubmit_live_flow_run 更好，因为它使用导致失败的确切数据。

defn = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
triggers = defn["properties"]["definition"]["triggers"]
manual = next(iter(triggers.values()))
print("Expected body:", manual.get("inputs", {}).get("schema"))

result = mcp("trigger_live_flow",
    environmentName=ENV, flowName=FLOW_ID,
    body={"name": "Test", "value": 1})
print(f"Status: {result['responseStatus']}, via: {result['invocation']}")
print(result.get("warning"))   # set when a required input was missing: the run still ran, with null

全新的非 HTTP 流程

全新的 Recurrence 流程无需变通方法：部署它，然后 立即使用 trigger_live_flow 且不带 body 运行它 —— 与门户中的 "Run flow" 按钮相同。body 会被拒绝；计划触发器不接受任何输入。

全新的连接器触发流程（SharePoint、webhook）没有先前 运行，并且没有真实的源事件就无法触发。使用临时的 HTTP 触发器部署，测试操作，然后切换到生产触发器：

production_trigger = definition["triggers"]
definition["triggers"] = {
    "manual": {"type": "Request", "kind": "Http", "inputs": {"schema": {}}}
}
result = mcp("update_live_flow", environmentName=ENV,
    flowName=FLOW_ID,       # omit if creating new
    definition=definition, connectionReferences=connection_references,
    displayName="Overdue Invoice Notifications")
FLOW_ID = FLOW_ID or result["flowName"]

mcp("trigger_live_flow", environmentName=ENV, flowName=FLOW_ID,
    body={"sample": "payload"})
runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=1)
if runs[0]["status"] == "Failed":
    err = mcp("get_live_flow_run_error",
        environmentName=ENV, flowName=FLOW_ID, runName=runs[0]["name"])
    raise Exception(err["failedActions"][-1])

definition["triggers"] = production_trigger
mcp("update_live_flow", environmentName=ENV, flowName=FLOW_ID,
    definition=definition, connectionReferences=connection_references)

触发器只是入口点；通过 HTTP 测试仍然会执行 相同的操作。如果操作使用 triggerBody() 或 triggerOutputs()，请传入一个 形状与生产触发器负载一致的代表性 body。

---

易踩问题

错误	后果	预防措施
部署时缺少 connectionReferences	400“Supply connectionReferences”	始终先调用 list_live_connections
Foreach 缺少 "operationOptions"	并行执行，写入时出现竞态条件	始终添加 "Sequential"
使用 union(old_data, new_data)	新值会覆盖旧值（优先采用第一个值）	使用 union(new_data, old_data)
对可能为 null 的字符串使用 split()	InvalidTemplate 崩溃	使用 coalesce(field, '') 包装
检查是否存在 result["error"]	该字段始终存在；真正的错误条件是 != null	使用 result.get("error") is not None
流已部署，但状态为“Stopped”	流不会按计划运行	使用 state: "Started" 调用 set_live_flow_state — 更改状态时不要使用 update_live_flow
Teams“Chat with Flow bot”的收件人使用对象	400 GraphUserDetailNotFound	使用末尾带分号的纯字符串（见下文）
Copilot/Skills 流不在解决方案中	Copilot Studio 可能无法将其识别为 Agent 工具	部署后，使用目标 solutionId 调用 add_live_flow_to_solution
使用 Button/Skills 触发器进行 MCP 测试	即使缺少必填输入（null），也会运行	在 trigger_live_flow 的 body 中传递输入；若出现 warning，请取消并使用完整的 body 重试
连接器操作缺少 metadata.operationMetadataId	设计器/仅运行 UI 的行为可能不一致	保留现有 ID；为新的连接器操作添加稳定的 GUID
Excel 占位 scriptId	保存时动态验证失败	部署前解析出真实的 Office Script ID
SharePoint PatchItem 未提供必填字段	即使字段未更改，也可能保存失败	回传未更改的必填字段，例如 item/Title
Copilot Studio 连接器调用草稿 Agent	连接器调用可能失败或触发过时行为	在测试/重新提交流之前，先发布该 Agent

Teams PostMessageToConversation — 收件人格式

body/recipient 参数格式取决于 location 的值：

位置	body/recipient 格式	示例
Chat with Flow bot	末尾带分号的纯电子邮件字符串	"[email protected];"
Channel	包含 groupId 和 channelId 的对象	{"groupId": "...", "channelId": "..."}

常见错误：为“Chat with Flow bot”传入 {"to": "[email protected]"} 会返回 400 GraphUserDetailNotFound 错误。API 要求使用纯字符串。

---

参考文件

• flow-schema.md — 完整的流定义 JSON 架构
• trigger-types.md — 触发器类型模板
• action-patterns-core.md — 变量、控制流和表达式
• action-patterns-data.md — 数组转换、HTTP 和解析
• action-patterns-connectors.md — SharePoint、Outlook、Teams 和 Approvals
• build-patterns.md — 完整的流定义模板（Recurrence+SP+Teams、HTTP 触发器）

相关技能

• power-automate-mcp — 核心连接设置和工具参考
• power-automate-debug — 部署后调试失败的流