Construire et déployer des flux Power Automate avec FlowStudio MCP

Guide étape par étape pour construire et déployer des flux cloud Power Automate par programmation via le serveur FlowStudio MCP.

Prérequis : Un serveur FlowStudio MCP doit être accessible avec un JWT valide. Consultez la skill power-automate-mcp pour la configuration de la connexion.
Inscrivez-vous sur https://mcp.flowstudio.app

Source de vérité

Appelez toujours tools/list en premier pour confirmer les noms d'outils disponibles et leurs schémas de paramètres. Les noms d'outils et les paramètres peuvent changer entre les versions du serveur. Cette skill couvre les formes de réponse, les notes de comportement et les modèles de construction — des choses que tools/list ne peut pas vous dire. Si ce document est en désaccord avec tools/list ou une vraie réponse API, l'API a raison.

Assistant 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

Étape 1 — Vérification de sécurité : le flux existe-t-il déjà ?

Vérifiez toujours avant de construire pour éviter les doublons :

results = mcp("list_live_flows", environmentName=ENV)

# list_live_flows returns { "flows": [...] }
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

Étape 2 — Obtenir les références de connexion

Chaque action de connecteur a besoin d'un connectionName qui pointe vers une clé dans la map connectionReferences du flux. Cette clé établit un lien vers une connexion authentifiée dans l'environnement.

OBLIGATOIRE : Vous DEVEZ appeler list_live_connections en premier — NE DEMANDEZ PAS à l'utilisateur des noms de connexion ou des GUID. L'API retourne les valeurs exactes dont vous avez besoin. Sollicitez l'utilisateur seulement si l'API confirme que les connexions requises manquent.

2a — Appelez toujours list_live_connections en premier

conns = 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()))

2b — Déterminez les connecteurs nécessaires au flux

En fonction du flux que vous construisez, identifiez les connecteurs requis. Noms d'API de connecteur courants :

Les flux qui n'ont besoin d'AUCUNE connexion (p. ex. Récurrence + Composition + HTTP uniquement) peuvent sauter le reste de l'Étape 2 — omettez connectionReferences de l'appel de déploiement.

2c — Si les connexions manquent, guidez l'utilisateur

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.

2d — Construire le bloc connectionReferences

Exécutez ceci seulement après que 2c confirme l'absence de connecteurs manquants :

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.connectionName dans les actions : Lors de la construction d'actions à l'Étape 3, définissez host.connectionName sur la clé de cette map (p. ex. shared_teams), NON pas le GUID de connexion. Le GUID va seulement à l'intérieur de l'entrée connectionReferences. Le moteur associe le host.connectionName de l'action à la clé pour trouver la bonne connexion.

Alternative — si vous avez déjà un flux utilisant les mêmes connecteurs, vous pouvez extraire connectionReferences de sa définition :

ref_flow = mcp("get_live_flow", environmentName=ENV, flowName="<existing-flow-id>")
connection_references = ref_flow["properties"]["connectionReferences"]

Consultez la référence connection-references.md de la skill power-automate-mcp pour la structure complète de la référence de connexion.

Étape 3 — Construire la définition du flux

Construisez l'objet de définition. Consultez flow-schema.md pour le schéma complet et ces références de modèles d'action pour les templates prêts à copier :

• action-patterns-core.md — Variables, flux de contrôle, expressions
• action-patterns-data.md — Transformations de tableau, HTTP, analyse
• 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
}

Consultez build-patterns.md pour des définitions de flux complètes et prêtes à l'emploi couvrant Récurrence+SharePoint+Teams, déclencheurs HTTP, et plus.

Étape 3a — Résoudre les valeurs dynamiques du connecteur

Quand une entrée d'action a besoin d'une valeur choisie dans un menu déroulant de connecteur (p. ex. un ID de liste SharePoint, un nom de table Dataverse, un UPN Azure AD d'utilisateur), utilisez get_live_dynamic_options pour le résoudre via MCP plutôt que de coder en dur des GUID.

# Resolve a SharePoint list by site
opts = mcp("get_live_dynamic_options",
    environmentName=ENV,
    connectorName="shared_sharepointonline",
    operationId="GetTables",
    parameters={"dataset": "https://contoso.sharepoint.com/sites/HR"})
# opts["value"] → [{"Name": "<list-guid>", "DisplayName": "Employees"}, ...]

Pont de paramètre externe automatique (serveur v1.1.6+) : vous pouvez passer des paramètres externes arbitraires directement dans parameters — le serveur synthétise maintenant le mappage parameterReference que PA's listEnum nécessite. Avant 1.1.6, vous deviez déclarer dynamicMetadata.parameters: {paramName: {parameterReference: "name"}} manuellement ou obtenir IncorrectDynamicInvokeParameter. Cela rend pratique d'invoquer des opérations de connecteur arbitraires via le pipeline dynamic-options (p. ex. shared_office365users.SearchUserV2 pour la recherche d'utilisateur AAD).

Secours du sélecteur d'utilisateur AadGraph

Pour les actions Outlook comme GetEmailsV3 (paramètres mailboxAddress, to, cc, from), PA's listEnum utilise builtInOperation:AadGraph.GetUsers — qui est cassé et retourne DynamicListValuesUndefinedOrInvalid pour chaque appel.

describe_live_connector (v1.1.6+) détecte ces paramètres et retourne un champ fallback structuré sur chaque paramètre affecté pointant vers une alternative qui fonctionne. Utilisez shared_office365users.SearchUserV2 pour résoudre la même forme d'utilisateur AAD {value: [{id, displayName, mail, userPrincipalName, ...}]} :

# Borrow a shared_office365users connection (any active one will do)
conn = next(c for c in conn_map if "office365users" in c)

users = mcp("get_live_dynamic_options",
    environmentName=ENV,
    connectorName="shared_office365users",
    connectionName=conn_map[conn],   # see Step 2a
    operationId="SearchUserV2",
    parameters={"searchTerm": "john", "top": 10})
# users["value"] → [{"Id": "...", "DisplayName": "John Smith", "Mail": "..."}, ...]

Insérez ensuite la valeur Mail résolue dans le paramètre de l'action Outlook — pas besoin d'appeler AadGraph.GetUsers directement.

Étape 4 — Déployer (créer ou mettre à jour)

update_live_flow gère à la fois la création et les mises à jour dans un seul outil.

Créer un nouveau flux (aucun flux existant)

Omettez flowName — le serveur génère un nouveau GUID et crée 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}")

Mettre à jour un flux existant

Fournissez flowName pour 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 retourne toujours une clé error. null (Python None) signifie le succès — ne traitez pas la présence de la clé comme un échec.

⚠️ description est obligatoire pour la création et la mise à jour.

Erreurs de déploiement courantes

Étape 5 — Vérifier le déploiement

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()))

Étape 6 — Tester le flux

OBLIGATOIRE : Avant de déclencher un test, demandez la confirmation de l'utilisateur. Exécuter un flux a de vrais effets secondaires — il peut envoyer des e-mails, poster des messages Teams, écrire dans SharePoint, démarrer des approbations ou appeler des API externes. Expliquez ce que le flux fera et attendez l'approbation explicite avant d'appeler trigger_live_flow ou resubmit_live_flow_run.

Flux mis à jour (ont des exécutions antérieures) — TOUT type de déclencheur

Utilisez resubmit_live_flow_run en premier. Cela fonctionne pour CHAQUE type de déclencheur — Récurrence, SharePoint, webhooks de connecteur, Bouton et HTTP. Il rejoue la charge utile de déclencheur originale. NE DEMANDEZ PAS à l'utilisateur de déclencher manuellement le flux ou d'attendre la prochaine exécution prévue.

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": "..."}

Flux déclenchés par HTTP — charge utile de test personnalisée

Utilisez trigger_live_flow seulement quand vous devez envoyer une charge utile différente de l'exécution originale. Pour vérifier une correction, resubmit_live_flow_run est meilleur car il utilise les données exactes qui ont causé l'échec.

schema = mcp("get_live_flow_http_schema",
    environmentName=ENV, flowName=FLOW_ID)
print("Expected body:", schema.get("requestSchema"))

result = mcp("trigger_live_flow",
    environmentName=ENV, flowName=FLOW_ID,
    body={"name": "Test", "value": 1})
print(f"Status: {result['responseStatus']}")

Flux non-HTTP tout neufs (Récurrence, déclencheurs de connecteur, etc.)

Un flux tout nouveau déclenché par Récurrence ou connecteur n'a aucune exécution antérieure à rejouer et aucun endpoint HTTP à appeler. C'est le SEUL scénario où vous avez besoin de l'approche de déclencheur HTTP temporaire ci-dessous. Déployez d'abord avec un déclencheur HTTP temporaire, testez les actions, puis basculez vers le déclencheur de production.

7a — Sauvegardez le déclencheur réel, déployez avec un déclencheur HTTP temporaire

# 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}")

7b — Déclenchez le flux et vérifiez le résultat

# 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

7c — Basculez vers le déclencheur de production

Une fois que l'exécution de test réussit, remplacez le déclencheur HTTP temporaire par le vrai :

# 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")

Pourquoi cela fonctionne : Le déclencheur est juste le point d'entrée — les actions sont identiques quel que soit la façon dont le flux démarre. Tester via un déclencheur HTTP exerce toutes les mêmes actions Composition, SharePoint, Teams, etc.

Déclencheurs de connecteur (p. ex. « Quand un élément est créé dans SharePoint ») : Si les actions référencent triggerBody() ou triggerOutputs(), passez une charge utile de test représentative dans le paramètre body de trigger_live_flow qui correspond à la forme que le déclencheur de connecteur produirait.

Pièges

Teams PostMessageToConversation — Formats de destinataire

Le format du paramètre body/recipient dépend de la valeur location :

Erreur courante : passer {"to": "user@contoso.com"} pour "Chat with Flow bot" retourne une erreur 400 GraphUserDetailNotFound. L'API attend une chaîne simple.

Fichiers de référence

• flow-schema.md — Schéma JSON de définition de flux complet
• trigger-types.md — Modèles de type de déclencheur
• action-patterns-core.md — Variables, flux de contrôle, expressions
• action-patterns-data.md — Transformations de tableau, HTTP, analyse
• action-patterns-connectors.md — SharePoint, Outlook, Teams, Approvals
• build-patterns.md — Modèles de définition de flux complets (Récurrence+SP+Teams, déclencheur HTTP)

Skills connexes

• power-automate-mcp — Skill de base : configuration de la connexion, assistant MCP, découverte d'outils
• power-automate-debug — Déboguer les flux qui échouent après le déploiement