Use Python with FlowRunner

This guide shows working code for the three ways Python and FlowRunner connect. Your script or service asks a person, and FlowRunner brings the answer back. A flow calls your Python service and waits for it to finish. A business event in an app starts your code, without another webhook receiver to host.

Your Python stays where it runs today, on cron, in a container, in Lambda, or behind FastAPI. FlowRunner doesn’t run Python. It calls your code over HTTP, or your code calls it. You build the flows from a chat with FlowRunner Prompt to Flow, so the FlowRunner side of each pattern is one request you paste.

1. Connect FlowRunner Prompt to Flow

FlowRunner Prompt to Flow is an MCP server you add to the AI tool you already use. You connect it once and sign in with your FlowRunner account. The server URL is the same for every tool: https://app.flowrunner.ai/mcp.

• Claude (web and desktop): open Settings > Connectors, click + Add, then Add custom connector. Enter FlowRunner and the server URL, click Add, then Connect, and approve the FlowRunner sign-in. On a Team or Enterprise plan, an Owner adds it once under Organization settings > Connectors.
• Claude Code: run these two commands in your project folder, and approve the sign-in page that opens:

claude mcp add --transport http flowrunner https://app.flowrunner.ai/mcp
claude mcp login flowrunner

• ChatGPT (on the web): open ChatGPT Plugins, click Add, then Add custom MCP server. Enter FlowRunner, put the server URL under Connection, choose OAuth, and approve the sign-in. In a chat, type @ and pick FlowRunner.
• Codex: add the server to ~/.codex/config.toml, then run codex mcp login flowrunner:

[mcp_servers.flowrunner]
url = "https://app.flowrunner.ai/mcp"

• Cursor: add the server to ~/.cursor/mcp.json, then open Customize > MCPs and follow the sign-in prompt:

{
  "mcpServers": {
    "flowrunner": { "url": "https://app.flowrunner.ai/mcp" }
  }
}

If you don't have a FlowRunner account yet, the Free plan needs no card, and every new account starts with 14 days of Professional. The setup guide in the docs has each tool step by step.

You’ll also need your workspace id and API key, from Workspace Settings > General > Credentials. Put them in FLOWRUNNER_WORKSPACE_ID and FLOWRUNNER_KEY on the machine that runs your code, and nowhere else. The key belongs to the whole workspace.

Your script asks a person, FlowRunner gets the answer

Say a nightly reconciliation script finds payments that don’t match their invoices. A person in finance has to decide each one, and that person doesn’t read your logs. The script hands each mismatch to a flow, the flow asks on Slack and waits, and the answer lands on your endpoint.

Build the flow. Paste a request like this into your AI tool with FlowRunner Prompt to Flow connected:

Build a flow called “Payment exception review” that starts from the Call Flow API. Its Initial Data is paymentId, invoiceId, expected, received, customer, and callbackUrl. Post a Slack message to #finance-review with those details and Approve and Reject buttons, then wait for the click. If nobody answers within 24 hours, send the same request by email to [email protected] and keep waiting. When the answer arrives, POST it to callbackUrl as JSON with paymentId, decision (approve or reject), any note the reviewer added, and this run’s Execution ID. Send an X-Callback-Secret header whose value I’ll paste into the block myself, and retry that POST on 5XX and network errors. Before you build, tell me anything you can’t map one to one, and ask me about anything I haven’t said.

Ask it to make the flow live when you’re happy with it, and note the flow id it reports.

Start it from Python. The script uses the non-blocking Call Flow API, which returns an executionId right away and leaves the run waiting for the reviewer:

import os

import requests

WORKSPACE = os.environ["FLOWRUNNER_WORKSPACE_ID"]
FLOW = "<flow-id>"  # Payment exception review
URL = f"https://api.flowrunner.ai/api/v1/{WORKSPACE}/flows/{FLOW}/activate"


def ask_for_review(mismatch: dict) -> str:
    resp = requests.post(
        URL,
        headers={"FR-Access-Key": os.environ["FLOWRUNNER_KEY"]},
        json={
            "paymentId": mismatch["payment_id"],
            "invoiceId": mismatch["invoice_id"],
            "expected": mismatch["expected"],
            "received": mismatch["received"],
            "customer": mismatch["customer"],
            "callbackUrl": os.environ["REVIEW_CALLBACK_URL"],
        },
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()["executionId"]

Store the executionId next to the payment, and the script is done for the night. requests sends the json= body with Content-Type: application/json, which the endpoint requires.

Receive the answer. The flow POSTs the decision to REVIEW_CALLBACK_URL. A small FastAPI endpoint checks the shared secret and acts on it:

import hmac
import os

from fastapi import FastAPI, Header, HTTPException

app = FastAPI()


@app.post("/reviews/answer")
def review_answer(body: dict, x_callback_secret: str = Header(default="")):
    if not hmac.compare_digest(x_callback_secret, os.environ["CALLBACK_SECRET"]):
        raise HTTPException(status_code=401)
    review = load_review(body["executionId"])  # what you stored when you started it
    if review.handled:
        return {"ok": True}  # a retried delivery
    if body["decision"] == "approve":
        post_adjustment(review, note=body.get("note", ""))
    else:
        flag_for_follow_up(review, note=body.get("note", ""))
    mark_handled(review)
    return {"ok": True}

Three things worth knowing:

• Don’t use the blocking call for this. Call Flow (Blocking) waits 5 seconds by default and 300 at most, which suits a lookup, not a person.
• Make the handler safe to run twice. The flow’s Retry Policy resends on 5XX, so a delivery your endpoint half-processed can arrive again. Matching on the executionId handles it.
• The wait is FlowRunner’s problem. The run can wait up to 30 days on Free, Starter, and Growth, and up to a year on Professional and above, and it’s still one execution. The Slack message, the reminder email, and the answer are all on the run record, which anyone on the team can open.

No web endpoint yet? A cron script can poll Checking a Run’s Status with the executionId and then read the decision block with Retrieving Block Results. Both take the same FR-Access-Key header.

FlowRunner calls your service

The other direction: your Python service does work nothing else can, and a flow calls it as one step among many. The HTTP Request block handles the call, and its Retry Policy reattempts on the status codes you list and on network errors, with fixed or exponential backoff.

Short calls are a single block. For work that takes minutes, have the service accept the job at once and report back when it’s done. The flow waits at an External Callback, and your service resumes that exact run.

Build the flow:

Build a flow called “Score new accounts” that runs every weekday at 07:00 UTC. Get the HubSpot contacts created in the last 24 hours. POST their ids as contactIds, with this run’s Execution ID as executionId, to https://scoring.example.com/score, with an Authorization header I’ll fill in with my bearer token in the flow editor. Retry on 429, 5XX, and network errors, three attempts, exponential backoff. Then wait at an External Callback for my service to report back with the scores. For each contact, update the HubSpot contact with the score, and send the owner a Slack message for any score over 80. If the scoring call fails after the retries, email me the error. Ask me about anything I haven’t said.

Accept the job, then report back. The service answers immediately and does the work in the background:

import os

import requests
from fastapi import BackgroundTasks, FastAPI

app = FastAPI()

RESUME_URL = (
    "https://api.flowrunner.ai/api/v1/"
    f"{os.environ['FLOWRUNNER_WORKSPACE_ID']}/flows/<flow-id>/triggers/<callback-block-id>/activate"
)


def score_and_report(execution_id: str, contact_ids: list[str]) -> None:
    scores = {cid: score_contact(cid) for cid in contact_ids}  # your model
    resp = requests.post(
        RESUME_URL,
        params={"execution": execution_id},
        headers={"FR-Access-Key": os.environ["FLOWRUNNER_KEY"]},
        json={"scores": scores},
        timeout=30,
    )
    resp.raise_for_status()


@app.post("/score")
def score(body: dict, background: BackgroundTasks):
    # check the Authorization header against your token here
    background.add_task(score_and_report, body["executionId"], body["contactIds"])
    return {"accepted": True}

The flow id and the callback block id are in the External Callback’s Callback URL, after /flow/ and /trigger/ (how to read them). The execution parameter tells FlowRunner which waiting run to resume. BackgroundTasks is fine for a job like this; for heavy work, FastAPI’s docs point you to a queue such as Celery, and the report-back call is the same from a worker.

Two other ways to let flows reach your code:

• A custom extension. If other people should use your service in their own flows, wrap its API in a custom extension: one JavaScript module deployed with the FlowRunner CLI, npm packages included, with its own blocks in the palette and its config on a form. npx flowrunner init-claude adds two Claude Code agents that write the extension and its tests. The service stays in Python.
• An MCP server. Expose your functions with the MCP Python SDK at a public URL and register it in FlowRunner, and an AI Agent block can call them as tools. The LangChain guide has an example.

A business event starts your code

Most scripts that react to the outside world need a receiver, a signature check, and an OAuth token per app. A flow can do that part. It starts from the app’s own trigger, with the account connected once in the workspace, and calls your endpoint with only the fields your code needs.

Say supplier invoices arrive as PDFs in a shared Gmail inbox, and you have a Python parser that knows your suppliers’ formats:

Build a flow that starts on a new Gmail attachment in the [email protected] inbox, only for PDF files. Save the attachment, then POST the sender, the subject, and a link to the saved file to https://parser.example.com/invoices, with an Authorization header I’ll fill in with my bearer token in the flow editor, and retry on 5XX. My service answers with vendor, invoiceNumber, amount, and dueDate. If the amount is over 5,000, send the details to the controller on Slack with Approve and Reject buttons and wait for the answer. Add approved invoices as a row in the “AP queue” Google Sheet, and reply to the sender when an invoice is rejected. Ask me about anything I haven’t said.

If the parser takes longer than a quick request, use the accept-and-report-back shape from the section above.

For jobs that run on a timer, a FlowRunner schedule can call your endpoint the same way. That’s worth it when the run history should sit where the rest of the team looks, or when a schedule and an app trigger should start the same code.

Optional: turn a script into a flow

Some scripts are mostly glue: read from one API, reshape, write to another, send a message. If you’d rather have that kind of script on the canvas, FlowRunner Prompt to Flow can read it and build the same steps as blocks. Paste the script and ask:

Read this Python script and build a FlowRunner flow that does the same thing with FlowRunner blocks and integrations. Keep the same order, conditions, and error handling. Tell me which parts are real computation that should stay in Python and be called over HTTP, and ask me about anything the code doesn’t make clear.

It’s an option, not the point of this guide. Code that does real work, a model, a parser, your own algorithm, belongs in Python, and the patterns above let FlowRunner call it where it lives.