flowstudio-power-automate-build by github/awesome-copilot
npx skills add https://github.com/github/awesome-copilot --skill flowstudio-power-automate-build通过 FlowStudio MCP 服务器以编程方式构建和部署 Power Automate 云流的逐步指南。
先决条件:必须能够访问一个具有有效 JWT 的 FlowStudio MCP 服务器。有关连接设置,请参阅 flowstudio-power-automate-mcp 技能。
订阅地址:https://mcp.flowstudio.app
务必首先调用
tools/list以确认可用的工具名称及其参数模式。工具名称和参数可能在服务器版本之间发生变化。本技能涵盖响应结构、行为说明和构建模式——这些是tools/list无法告诉你的信息。如果本文档与tools/list或实际 API 响应不一致,以 API 为准。
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>" # 例如:Default-xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
广告位招租
在这里展示您的产品或服务
触达数万 AI 开发者,精准高效
构建前务必先检查,避免重复:
results = mcp("list_store_flows",
environmentName=ENV, searchTerm="My New Flow")
# list_store_flows 返回一个直接数组(无包装对象)
if len(results) > 0:
# 流已存在 — 进行修改而非创建
# id 格式为 "envId.flowId" — 分割以获取流 UUID
FLOW_ID = results[0]["id"].split(".", 1)[1]
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
每个连接器操作都需要一个 connectionName,它指向流 connectionReferences 映射中的一个键。该键链接到环境中的一个已认证连接。
强制要求:你必须首先调用
list_live_connections— 不要向用户询问连接名称或 GUID。API 会返回你需要的精确值。仅当 API 确认缺少必需连接时,才提示用户。
list_live_connectionsconns = mcp("list_live_connections", environmentName=ENV)
# 仅筛选已连接(已认证)的连接
active = [c for c in conns["connections"]
if c["statuses"][0]["status"] == "Connected"]
# 构建查找表:connectorName → connectionName (id)
conn_map = {}
for c in active:
conn_map[c["connectorName"]] = c["id"]
print(f"Found {len(active)} active connections")
print("Available connectors:", list(conn_map.keys()))
根据你正在构建的流,确定需要哪些连接器。常见的连接器 API 名称:
| 连接器 | API 名称 |
|---|---|
| SharePoint | shared_sharepointonline |
| Outlook / Office 365 | shared_office365 |
| Teams | shared_teams |
| Approvals | shared_approvals |
| OneDrive for Business | shared_onedriveforbusiness |
| Excel Online (Business) | shared_excelonlinebusiness |
| Dataverse | shared_commondataserviceforapps |
| Microsoft Forms | shared_microsoftforms |
不需要连接 的流(例如,仅包含 Recurrence + Compose + HTTP)可以跳过步骤 2 的其余部分 — 在部署调用中省略
connectionReferences。
connectors_needed = ["shared_sharepointonline", "shared_office365"] # 根据流调整
missing = [c for c in connectors_needed if c not in conn_map]
if not missing:
print("✅ All required connections are available — proceeding to build")
else:
# ── 停止:连接必须交互式创建 ──
# 连接需要在浏览器中进行 OAuth 授权 — 没有 API 可以创建它们。
print("⚠️ The following connectors have no active connection in this environment:")
for c in missing:
friendly = c.replace("shared_", "").replace("onlinebusiness", " Online (Business)")
print(f" • {friendly} (API name: {c})")
print()
print("Please create the missing connections:")
print(" 1. Open https://make.powerautomate.com/connections")
print(" 2. Select the correct environment from the top-right picker")
print(" 3. Click '+ New connection' for each missing connector listed above")
print(" 4. Sign in and authorize when prompted")
print(" 5. Tell me when done — I will re-check and continue building")
# 在用户确认之前,不要继续步骤 3。
# 用户确认后,重新运行步骤 2a 以刷新 conn_map。
仅在 2c 确认没有缺少的连接器后执行此步骤:
connection_references = {}
for connector in connectors_needed:
connection_references[connector] = {
"connectionName": conn_map[connector], # 来自 list_live_connections 的 GUID
"source": "Invoker",
"id": f"/providers/Microsoft.PowerApps/apis/{connector}"
}
重要 — 操作中的
host.connectionName:在步骤 3 中构建操作时,将host.connectionName设置为此映射中的键(例如shared_teams),而不是连接 GUID。GUID 仅放在connectionReferences条目内部。引擎将操作的host.connectionName与键匹配以找到正确的连接。
替代方案 — 如果你已经有一个使用相同连接器的流,可以从其定义中提取
connectionReferences:ref_flow = mcp("get_live_flow", environmentName=ENV, flowName="<existing-flow-id>") connection_references = ref_flow["properties"]["connectionReferences"]
有关完整的连接引用结构,请参阅 power-automate-mcp 技能的 connection-references.md 参考文档。
构建定义对象。完整模式请参阅 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": { ... }, # 参见 trigger-types.md / build-patterns.md "actions": { ... } # 参见 ACTION-PATTERNS-*.md / build-patterns.md }
完整、可直接使用的流定义(涵盖 Recurrence+SharePoint+Teams、HTTP 触发器等)请参阅 build-patterns.md。
update_live_flow 在一个工具中处理创建和更新。
省略 flowName — 服务器生成新的 GUID 并通过 PUT 创建:
result = mcp("update_live_flow",
environmentName=ENV,
# 省略 flowName → 创建新流
definition=definition,
connectionReferences=connection_references,
displayName="Overdue Invoice Notifications",
description="Weekly SharePoint → Teams notification flow, built by agent"
)
if result.get("error") is not None:
print("Create failed:", result["error"])
else:
# 捕获新流 ID 供后续步骤使用
FLOW_ID = result["created"]
print(f"✅ Flow created: {FLOW_ID}")
提供 flowName 以进行 PATCH:
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID,
definition=definition,
connectionReferences=connection_references,
displayName="My Updated Flow",
description="Updated by agent on " + __import__('datetime').datetime.utcnow().isoformat()
)
if result.get("error") is not None:
print("Update failed:", result["error"])
else:
print("Update succeeded:", result)
⚠️
update_live_flow总是返回一个error键。null(Python 中的None)表示成功 — 不要将该键的存在视为失败。⚠️ 创建和更新都需要
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 |
check = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
# 确认状态
print("State:", check["properties"]["state"]) # 应为 "Started"
# 确认我们添加的操作存在
acts = check["properties"]["definition"]["actions"]
print("Actions:", list(acts.keys()))
强制要求:在触发任何测试运行之前,务必请求用户确认。运行流会产生真实的副作用 — 它可能会发送电子邮件、发布 Teams 消息、写入 SharePoint、启动审批或调用外部 API。解释流将执行的操作,并在调用
trigger_live_flow或resubmit_live_flow_run之前等待明确批准。
最快路径 — 重新提交最近的运行:
runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=1)
if runs:
result = mcp("resubmit_live_flow_run",
environmentName=ENV, flowName=FLOW_ID, runName=runs[0]["name"])
print(result)
直接使用测试负载触发:
schema = mcp("get_live_flow_http_schema",
environmentName=ENV, flowName=FLOW_ID)
print("Expected body:", schema.get("triggerSchema"))
result = mcp("trigger_live_flow",
environmentName=ENV, flowName=FLOW_ID,
body={"name": "Test", "value": 1})
print(f"Status: {result['status']}")
全新的 Recurrence 或连接器触发的流没有可重新提交的运行,也没有可调用的 HTTP 端点。首先使用临时的 HTTP 触发器部署,测试操作,然后切换到生产触发器。
# 保存你在步骤 3 中构建的生产触发器
production_trigger = definition["triggers"]
# 替换为临时 HTTP 触发器
definition["triggers"] = {
"manual": {
"type": "Request",
"kind": "Http",
"inputs": {
"schema": {}
}
}
}
# 使用临时触发器部署(创建或更新)
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID, # 如果是创建新流则省略
definition=definition,
connectionReferences=connection_references,
displayName="Overdue Invoice Notifications",
description="Deployed with temp HTTP trigger for testing")
if result.get("error") is not None:
print("Deploy failed:", result["error"])
else:
if not FLOW_ID:
FLOW_ID = result["created"]
print(f"✅ Deployed with temp HTTP trigger: {FLOW_ID}")
# 触发流
test = mcp("trigger_live_flow",
environmentName=ENV, flowName=FLOW_ID)
print(f"Trigger response status: {test['status']}")
# 等待运行完成
import time; time.sleep(15)
# 检查运行结果
runs = mcp("get_live_flow_runs",
environmentName=ENV, flowName=FLOW_ID, top=1)
run = runs[0]
print(f"Run {run['name']}: {run['status']}")
if run["status"] == "Failed":
err = mcp("get_live_flow_run_error",
environmentName=ENV, flowName=FLOW_ID, runName=run["name"])
root = err["failedActions"][-1]
print(f"Root cause: {root['actionName']} → {root.get('code')}")
# 在继续之前调试并修复定义
# 完整诊断工作流请参阅 power-automate-debug 技能
测试运行成功后,将临时 HTTP 触发器替换为真实的触发器:
# 恢复生产触发器
definition["triggers"] = production_trigger
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID,
definition=definition,
connectionReferences=connection_references,
description="Swapped to production trigger after successful test")
if result.get("error") is not None:
print("Trigger swap failed:", result["error"])
else:
print("✅ Production trigger deployed — flow is live")
为什么这可行:触发器只是入口点 — 无论流如何启动,操作都是相同的。通过 HTTP 触发器测试会执行所有相同的 Compose、SharePoint、Teams 等操作。
连接器触发器(例如“在 SharePoint 中创建项目时”):如果操作引用了
triggerBody()或triggerOutputs(),请在trigger_live_flow的body参数中传递一个具有代表性的测试负载,该负载应与连接器触发器产生的形状匹配。
| 错误 | 后果 | 预防措施 |
|---|---|---|
部署中缺少 connectionReferences | 400 "Supply connectionReferences" | 务必首先调用 list_live_connections |
Foreach 上缺少 "operationOptions" | 并行执行,写入时出现竞态条件 | 始终添加 "Sequential" |
union(old_data, new_data) | 旧值覆盖新值(先到先得) | 使用 union(new_data, old_data) |
对可能为空的字符串使用 split() | InvalidTemplate 崩溃 | 使用 coalesce(field, '') 包装 |
检查 result["error"] 是否存在 | 始终存在;真正的错误是 != null | 使用 result.get("error") is not None |
| 流已部署但状态为 "Stopped" | 流不会按计划运行 | 检查连接授权;重新启用 |
| Teams "与 Flow 机器人聊天" 收件人格式为对象 | 400 GraphUserDetailNotFound | 使用带尾随分号的纯字符串(见下文) |
PostMessageToConversation — 收件人格式body/recipient 参数格式取决于 location 值:
| 位置 | body/recipient 格式 | 示例 |
|---|---|---|
| 与 Flow 机器人聊天 | 带尾随分号的纯电子邮件字符串 | "user@contoso.com;" |
| 频道 | 包含 groupId 和 channelId 的对象 | {"groupId": "...", "channelId": "..."} |
常见错误:为“与 Flow 机器人聊天”传递
{"to": "user@contoso.com"}会返回 400GraphUserDetailNotFound错误。API 期望一个纯字符串。
flowstudio-power-automate-mcp — 核心连接设置和工具参考flowstudio-power-automate-debug — 部署后调试失败的流每周安装数
491
代码库
GitHub 星标数
26.7K
首次出现
2026年3月8日
安全审计
安装于
gemini-cli437
codex436
opencode425
cursor422
github-copilot419
kimi-cli418
Step-by-step guide for constructing and deploying Power Automate cloud flows programmatically through the FlowStudio MCP server.
Prerequisite : A FlowStudio MCP server must be reachable with a valid JWT. See the flowstudio-power-automate-mcp skill for connection setup.
Subscribe at https://mcp.flowstudio.app
Always call
tools/listfirst to confirm available tool names and their parameter schemas. Tool names and parameters may change between server versions. This skill covers response shapes, behavioral notes, and build patterns — thingstools/listcannot tell you. If this document disagrees withtools/listor a real API response, the API wins.
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
Always look before you build to avoid duplicates:
results = mcp("list_store_flows",
environmentName=ENV, searchTerm="My New Flow")
# list_store_flows returns a direct array (no wrapper object)
if len(results) > 0:
# Flow exists — modify rather than create
# id format is "envId.flowId" — split to get the flow UUID
FLOW_ID = results[0]["id"].split(".", 1)[1]
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
Every connector action needs a connectionName that points to a key in the flow's connectionReferences map. That key links to an authenticated connection in the environment.
MANDATORY : You MUST call
list_live_connectionsfirst — do NOT ask the user for connection names or GUIDs. The API returns the exact values you need. Only prompt the user if the API confirms that required connections are missing.
list_live_connections firstconns = mcp("list_live_connections", environmentName=ENV)
# Filter to connected (authenticated) connections only
active = [c for c in conns["connections"]
if c["statuses"][0]["status"] == "Connected"]
# Build a lookup: connectorName → connectionName (id)
conn_map = {}
for c in active:
conn_map[c["connectorName"]] = c["id"]
print(f"Found {len(active)} active connections")
print("Available connectors:", list(conn_map.keys()))
Based on the flow you are building, identify which connectors are required. Common connector API names:
| Connector | API name |
|---|---|
| SharePoint | shared_sharepointonline |
| Outlook / Office 365 | shared_office365 |
| Teams | shared_teams |
| Approvals | shared_approvals |
| OneDrive for Business | shared_onedriveforbusiness |
| Excel Online (Business) | shared_excelonlinebusiness |
Flows that need NO connections (e.g. Recurrence + Compose + HTTP only) can skip the rest of Step 2 — omit
connectionReferencesfrom the deploy call.
connectors_needed = ["shared_sharepointonline", "shared_office365"] # adjust per flow
missing = [c for c in connectors_needed if c not in conn_map]
if not missing:
print("✅ All required connections are available — proceeding to build")
else:
# ── STOP: connections must be created interactively ──
# Connections require OAuth consent in a browser — no API can create them.
print("⚠️ The following connectors have no active connection in this environment:")
for c in missing:
friendly = c.replace("shared_", "").replace("onlinebusiness", " Online (Business)")
print(f" • {friendly} (API name: {c})")
print()
print("Please create the missing connections:")
print(" 1. Open https://make.powerautomate.com/connections")
print(" 2. Select the correct environment from the top-right picker")
print(" 3. Click '+ New connection' for each missing connector listed above")
print(" 4. Sign in and authorize when prompted")
print(" 5. Tell me when done — I will re-check and continue building")
# DO NOT proceed to Step 3 until the user confirms.
# After user confirms, re-run Step 2a to refresh conn_map.
Only execute this after 2c confirms no missing connectors:
connection_references = {}
for connector in connectors_needed:
connection_references[connector] = {
"connectionName": conn_map[connector], # the GUID from list_live_connections
"source": "Invoker",
"id": f"/providers/Microsoft.PowerApps/apis/{connector}"
}
IMPORTANT —
host.connectionNamein actions: When building actions in Step 3, sethost.connectionNameto the key from this map (e.g.shared_teams), NOT the connection GUID. The GUID only goes inside theconnectionReferencesentry. The engine matches the action'shost.connectionNameto the key to find the right connection.
Alternative — if you already have a flow using the same connectors, you can extract
connectionReferencesfrom its definition:ref_flow = mcp("get_live_flow", environmentName=ENV, flowName="<existing-flow-id>") connection_references = ref_flow["properties"]["connectionReferences"]
See the power-automate-mcp skill's connection-references.md reference for the full connection reference structure.
Construct the definition object. See flow-schema.md for the full schema and these action pattern references for copy-paste templates:
action-patterns-core.md — Variables, control flow, expressions
action-patterns-data.md — Array transforms, HTTP, parsing
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 }
See build-patterns.md for complete, ready-to-use flow definitions covering Recurrence+SharePoint+Teams, HTTP triggers, and more.
update_live_flow handles both creation and updates in a single tool.
Omit flowName — the server generates a new GUID and creates via PUT:
result = mcp("update_live_flow",
environmentName=ENV,
# flowName omitted → creates a new flow
definition=definition,
connectionReferences=connection_references,
displayName="Overdue Invoice Notifications",
description="Weekly SharePoint → Teams notification flow, built by agent"
)
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}")
Provide flowName to PATCH:
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID,
definition=definition,
connectionReferences=connection_references,
displayName="My Updated Flow",
description="Updated by agent on " + __import__('datetime').datetime.utcnow().isoformat()
)
if result.get("error") is not None:
print("Update failed:", result["error"])
else:
print("Update succeeded:", result)
⚠️
update_live_flowalways returns anerrorkey.null(PythonNone) means success — do not treat the presence of the key as failure.⚠️
descriptionis required for both create and update.
| Error message (contains) | Cause | Fix |
|---|---|---|
missing from connectionReferences | An action's host.connectionName references a key that doesn't exist in the connectionReferences map | Ensure host.connectionName uses the key from connectionReferences (e.g. shared_teams), not the raw GUID |
ConnectionAuthorizationFailed / 403 | The connection GUID belongs to another user or is not authorized |
check = mcp("get_live_flow", environmentName=ENV, flowName=FLOW_ID)
# Confirm state
print("State:", check["properties"]["state"]) # Should be "Started"
# Confirm the action we added is there
acts = check["properties"]["definition"]["actions"]
print("Actions:", list(acts.keys()))
MANDATORY : Before triggering any test run, ask the user for confirmation. Running a flow has real side effects — it may send emails, post Teams messages, write to SharePoint, start approvals, or call external APIs. Explain what the flow will do and wait for explicit approval before calling
trigger_live_floworresubmit_live_flow_run.
The fastest path — resubmit the most recent run:
runs = mcp("get_live_flow_runs", environmentName=ENV, flowName=FLOW_ID, top=1)
if runs:
result = mcp("resubmit_live_flow_run",
environmentName=ENV, flowName=FLOW_ID, runName=runs[0]["name"])
print(result)
Fire directly with a test payload:
schema = mcp("get_live_flow_http_schema",
environmentName=ENV, flowName=FLOW_ID)
print("Expected body:", schema.get("triggerSchema"))
result = mcp("trigger_live_flow",
environmentName=ENV, flowName=FLOW_ID,
body={"name": "Test", "value": 1})
print(f"Status: {result['status']}")
A brand-new Recurrence or connector-triggered flow has no runs to resubmit and no HTTP endpoint to call. Deploy with a temporary HTTP trigger first, test the actions, then swap to the production trigger.
# Save the production trigger you built in Step 3
production_trigger = definition["triggers"]
# Replace with a temporary HTTP trigger
definition["triggers"] = {
"manual": {
"type": "Request",
"kind": "Http",
"inputs": {
"schema": {}
}
}
}
# Deploy (create or update) with the temp trigger
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID, # omit if creating new
definition=definition,
connectionReferences=connection_references,
displayName="Overdue Invoice Notifications",
description="Deployed with temp HTTP trigger for testing")
if result.get("error") is not None:
print("Deploy failed:", result["error"])
else:
if not FLOW_ID:
FLOW_ID = result["created"]
print(f"✅ Deployed with temp HTTP trigger: {FLOW_ID}")
# Trigger the flow
test = mcp("trigger_live_flow",
environmentName=ENV, flowName=FLOW_ID)
print(f"Trigger response status: {test['status']}")
# Wait for the run to complete
import time; time.sleep(15)
# Check the run result
runs = mcp("get_live_flow_runs",
environmentName=ENV, flowName=FLOW_ID, top=1)
run = runs[0]
print(f"Run {run['name']}: {run['status']}")
if run["status"] == "Failed":
err = mcp("get_live_flow_run_error",
environmentName=ENV, flowName=FLOW_ID, runName=run["name"])
root = err["failedActions"][-1]
print(f"Root cause: {root['actionName']} → {root.get('code')}")
# Debug and fix the definition before proceeding
# See power-automate-debug skill for full diagnosis workflow
Once the test run succeeds, replace the temporary HTTP trigger with the real one:
# Restore the production trigger
definition["triggers"] = production_trigger
result = mcp("update_live_flow",
environmentName=ENV,
flowName=FLOW_ID,
definition=definition,
connectionReferences=connection_references,
description="Swapped to production trigger after successful test")
if result.get("error") is not None:
print("Trigger swap failed:", result["error"])
else:
print("✅ Production trigger deployed — flow is live")
Why this works : The trigger is just the entry point — the actions are identical regardless of how the flow starts. Testing via HTTP trigger exercises all the same Compose, SharePoint, Teams, etc. actions.
Connector triggers (e.g. "When an item is created in SharePoint"): If actions reference
triggerBody()ortriggerOutputs(), pass a representative test payload intrigger_live_flow'sbodyparameter that matches the shape the connector trigger would produce.
| Mistake | Consequence | Prevention |
|---|---|---|
Missing connectionReferences in deploy | 400 "Supply connectionReferences" | Always call list_live_connections first |
"operationOptions" missing on Foreach | Parallel execution, race conditions on writes | Always add "Sequential" |
union(old_data, new_data) | Old values override new (first-wins) | Use union(new_data, old_data) |
PostMessageToConversation — Recipient FormatsThe body/recipient parameter format depends on the location value:
| Location | body/recipient format | Example |
|---|---|---|
| Chat with Flow bot | Plain email string with trailing semicolon | "user@contoso.com;" |
| Channel | Object with groupId and channelId | {"groupId": "...", "channelId": "..."} |
Common mistake : passing
{"to": "user@contoso.com"}for "Chat with Flow bot" returns a 400GraphUserDetailNotFounderror. The API expects a plain string.
flowstudio-power-automate-mcp — Core connection setup and tool referenceflowstudio-power-automate-debug — Debug failing flows after deploymentWeekly Installs
491
Repository
GitHub Stars
26.7K
First Seen
Mar 8, 2026
Security Audits
Gen Agent Trust HubPassSocketPassSnykWarn
Installed on
gemini-cli437
codex436
opencode425
cursor422
github-copilot419
kimi-cli418
agent-browser 浏览器自动化工具 - Vercel Labs 命令行网页操作与测试
136,300 周安装
退货与逆向物流管理指南:策略、分级、处置与欺诈检测全流程解析
520 周安装
学术研究写作助手 - AI驱动的学术论文写作工具,支持IEEE格式引用与同行评审来源
521 周安装
fal.ai AI图像编辑工具:风格迁移、对象移除、背景更换一键处理
521 周安装
skill-name 技能使用指南 - 快速上手、命令示例与故障排除
521 周安装
OpenAI DOCX技能:专业文档处理与格式渲染指南 | Python-docx、LibreOffice
522 周安装
Apple Notes CLI 终端命令行工具 - 在 macOS 终端管理 Apple 笔记
522 周安装
| Dataverse | shared_commondataserviceforapps |
| Microsoft Forms | shared_microsoftforms |
Re-run Step 2a and use a connection owned by the current x-api-key user |
InvalidTemplate / InvalidDefinition | Syntax error in the definition JSON | Check runAfter chains, expression syntax, and action type spelling |
ConnectionNotConfigured | A connector action exists but the connection GUID is invalid or expired | Re-check list_live_connections for a fresh GUID |
split() on potentially-null string | InvalidTemplate crash | Wrap with coalesce(field, '') |
Checking result["error"] exists | Always present; true error is != null | Use result.get("error") is not None |
| Flow deployed but state is "Stopped" | Flow won't run on schedule | Check connection auth; re-enable |
| Teams "Chat with Flow bot" recipient as object | 400 GraphUserDetailNotFound | Use plain string with trailing semicolon (see below) |