Construir e Implantar Fluxos do Power Automate com FlowStudio MCP

Guia passo a passo para construir e implantar fluxos de nuvem do Power Automate programaticamente através do servidor FlowStudio MCP.

Pré-requisito: Um servidor FlowStudio MCP deve ser acessível com um JWT válido. Consulte a habilidade flowstudio-power-automate-mcp para configuração de conexão. Assine em https://mcp.flowstudio.app

Fluxo de trabalho:

1. Carregar ferramentas de build atuais.
2. Verificar se existe um fluxo.
3. Resolver referências de conexão.
4. Construir a definição.
5. Implantar.
6. Verificar.
7. Testar.

Fonte da Verdade

Sempre chame list_skills / tool_search primeiro para confirmar os nomes das ferramentas disponíveis e os esquemas de parâmetros. Nomes de ferramentas e parâmetros podem mudar entre versões do servidor. Esta habilidade abrange formas de resposta, notas de comportamento e padrões de construção — coisas que os esquemas de ferramentas não podem lhe dizer. Se este documento discordar de tool_search ou de uma resposta real da API, a API prevalece.

Auxiliar 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. Carregar as Ferramentas de Build Atuais

Para um fluxo totalmente novo, carregue o pacote create-flow do servidor. Para editar um fluxo existente, carregue build-flow. Isso mantém o agente alinhado com o esquema atual do servidor MCP antes de construir o JSON.

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.

Se você precisar de uma ferramenta fora do pacote, carregue-a explicitamente:

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

1. Verificação de Segurança: O Fluxo Já Existe?

Sempre verifique antes de construir para evitar duplicatas:

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

Para ambientes muito grandes, list_live_flows pode retornar um URL de continuação. Passe-o de volta como continuationUrl com o mesmo mode para recuperar o próximo lote. Use mode="admin" apenas quando o usuário precisar de todos os fluxos do ambiente e a identidade MCP tiver direitos de administrador.

2. Obter Referências de Conexão

Cada ação do conector precisa de uma connectionName que aponte para uma chave no mapa connectionReferences do fluxo. Essa chave se vincula a uma conexão autenticada no ambiente.

OBRIGATÓRIO: VOCÊ DEVE chamar list_live_connections primeiro — NÃO pergunte ao usuário nomes ou GUIDs de conexão. A API retorna os valores exatos de que você precisa. Só solicite ao usuário se a API confirmar que as conexões necessárias estão ausentes.

2a — Encontrar conexões ativas

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}

Para um conector conhecido, passe search para reduzir a saída e obter valores prontos para copiar de connectionReferenceTemplate e hostTemplate:

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

…