Power Automate × Python── Where Python fits into flows (desktop and Azure Functions)

# Run Python script (Power Automate for desktop). Set "Python version" to Python 3.4.
# Keep to old syntax: no f-strings; use str.format.
import csv
import io
import os
from collections import Counter

ledger_path = r"%LedgerPath%"  # replaced by the flow variable at run time
if not os.path.exists(ledger_path):
    ledger_path = "samples/ledger.csv"  # local test outside the flow

with io.open(ledger_path, encoding="utf-8-sig", newline="") as f:
    rows = list(csv.DictReader(f))

status = Counter(r["状態"] for r in rows)
open_rows = [r for r in rows if r["状態"] != "完了"]
print("total={0}".format(len(rows)))
print("open={0}".format(len(open_rows)))
for name, n in sorted(status.items()):
    print("status.{0}={1}".format(name, n))

"""HTTP-triggered function: take ledger rows as JSON and return counts."""
import json
from collections import Counter

import azure.functions as func

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)


def summarize_rows(rows):
    status = Counter(r.get("状態", "") for r in rows)
    open_by_owner = Counter(r.get("担当者", "") for r in rows if r.get("状態") != "完了")
    return {"total": len(rows), "status": dict(status), "open_by_owner": dict(open_by_owner)}


@app.route(route="summarize", methods=["POST"])
def summarize(req: func.HttpRequest) -> func.HttpResponse:
    try:
        rows = req.get_json().get("rows", [])
    except ValueError:
        return func.HttpResponse("JSON body with 'rows' is required", status_code=400)
    body = json.dumps(summarize_rows(rows), ensure_ascii=False)
    return func.HttpResponse(body, mimetype="application/json")

POST /api/summarize HTTP/1.1
Host: <APP_NAME>.azurewebsites.net
Content-Type: application/json
x-functions-key: <FUNCTION_KEY>

{"rows": [{"依頼ID": "R-2026-002", "状態": "受付", "担当者": "担当A"},
          {"依頼ID": "R-2026-004", "状態": "完了", "担当者": "担当A"}]}

"""Send JSON to a flow that starts with 'When an HTTP request is received' (DRY_RUN by default)."""
import json
import os

import requests

DRY_RUN = os.environ.get("DRY_RUN", "1") != "0"
FLOW_URL = os.environ.get("FLOW_URL", "")  # carries a SAS signature: keep it out of code and logs
BEARER = os.environ.get("FLOW_BEARER_TOKEN", "")  # needed when the trigger allows tenant users only


def build_request(rows):
    headers = {"Content-Type": "application/json"}
    if BEARER:
        headers["Authorization"] = "Bearer " + BEARER
    return headers, {"source": "python", "rows": rows}


def main():
    rows = [{"依頼ID": "R-2026-041", "文書名": "文書41（架空）", "状態": "受付"}]
    headers, body = build_request(rows)
    if DRY_RUN or not FLOW_URL:
        print("DRY_RUN: would POST", len(rows), "row(s); bearer token:", "Authorization" in headers)
        print(json.dumps(body, ensure_ascii=False))
        return
    r = requests.post(FLOW_URL, headers=headers, json=body, timeout=30)
    print("status:", r.status_code)


if __name__ == "__main__":
    main()

// Office Script (TypeScript), called from Power Automate "Run script".
// Counts rows whose status is not "完了" and returns the number to the flow.
function main(workbook: ExcelScript.Workbook, statusColumn: string): number {
  const table = workbook.getTable("RequestLedger");
  const values = table.getColumnByName(statusColumn).getRangeBetweenHeaderAndTotal().getValues();
  let open = 0;
  values.forEach((row) => {
    if (row[0] !== "完了") {
      open += 1;
    }
  });
  return open;
}

{
  "swagger": "2.0",
  "info": {
    "version": "1.0.0",
    "title": "LedgerSummary",
    "description": "Summarize request-ledger rows with an Azure Function"
  },
  "host": "example-func.azurewebsites.net",
  "basePath": "/api",
  "schemes": [
    "https"
  ],
  "consumes": [
    "application/json"
  ],
  "produces": [
    "application/json"
  ],
  "securityDefinitions": {
    "function_key": {
      "type": "apiKey",
      "in": "header",
      "name": "x-functions-key"
    }
  },
  "security": [
    {
      "function_key": []
    }
  ],
  "paths": {
    "/summarize": {
      "post": {
        "summary": "Count ledger rows by status",
        "operationId": "SummarizeLedger",
        "parameters": [
          {
            "in": "body",
            "name": "body",
            "schema": {
              "type": "object",
              "properties": {
                "rows": {
                  "type": "array",
                  "items": {
                    "type": "object"
                  },
                  "x-ms-summary": "rows"
                }
              }
            }
          }
        ],
        "responses": {
          "200": {
            "description": "200",
            "schema": {
              "type": "object",
              "properties": {
                "total": {
                  "type": "integer",
                  "x-ms-summary": "total"
                },
                "status": {
                  "type": "object",
                  "x-ms-summary": "status"
                },
                "open_by_owner": {
                  "type": "object",
                  "x-ms-summary": "open by owner"
                }
              }
            }
          }
        }
      }
    }
  }
}

@app.activity_trigger(input_name="order")
def prepare_order_activity(order: dict) -> dict:
    # Validation and data minimization belong in an activity
    return {
        "order_id": order["order_id"],
        "customer_id": order["customer"]["id"],
        "items": order["items"],
    }


@app.orchestration_trigger(context_name="context")
def order_orchestrator(context: DurableAgentContext):
    prepared_order = yield context.call_activity(
        "prepare_order_activity",
        context.get_input(),
    )
    # The model call is scheduled as an extension-managed activity
    assessment = yield context.call_agent(
        "order-fulfillment",
        {"order": prepared_order, "task": "assess fulfillment risk"},
    )
    return {
        "order_id": prepared_order["order_id"],
        "risk_assessment": assessment,
    }

# tools/incident_tools.py
from typing import Any
from azure_functions_agents import workflow_tool

@workflow_tool(description="Fetch recent log lines for a service.")
async def fetch_logs(args: dict[str, Any]) -> dict[str, Any]:
    service = args["service"]
    return {"service": service, "lines": ["..."]}

# Power Automate replaces %...% with the variable value before the script runs
variableName = "%LedgerText%"
rowCount = len(variableName.splitlines())
# Return the result to the flow with print
print(rowCount)