Use LangChain with FlowRunner

This guide connects a LangChain agent to FlowRunner in three ways. Your agent calls a FlowRunner flow as a tool, so one tool call runs a whole business process. Your agent’s HumanInTheLoopMiddleware pauses, and a flow gets the decision from a person in Slack or email. And FlowRunner calls your agent, from a flow or from a FlowRunner AI agent, over HTTP or MCP.

Your agent stays in your service, on whatever runs it today. FlowRunner doesn’t run Python. The code below is LangChain v1 with create_agent, plus requests and FastAPI for the HTTP parts, and the flows are built from a chat with FlowRunner Prompt to Flow.

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.

Your code needs the workspace id and API key, from Workspace Settings > General > Credentials. Keep them in FLOWRUNNER_WORKSPACE_ID and FLOWRUNNER_KEY on the server that runs the agent. The key is sent in the FR-Access-Key header, never in a URL.

Give your agent a flow as a tool

A sales agent that can “send a quote” needs a lot behind that phrase: find or create the customer in QuickBooks, create the estimate, send it, tell the account owner. Writing that as four tools means four sets of credentials and four places to fail. As one flow, it’s one tool.

Build the flow:

Build a flow called “Send quote” that starts from the Call Flow API. Its Initial Data is customerEmail, customerName, lineItems (a list of name, quantity, and price), and callbackUrl. Find the customer in QuickBooks by email, or create them. Create an estimate with the line items and send it to the customer. Post the estimate number and total to #sales on Slack. Then POST estimateNumber, total, and this run’s Execution ID to callbackUrl with an X-Callback-Secret header whose value I’ll paste into the block myself, and retry on 5XX. If any QuickBooks step fails, post the error to #sales-ops instead. Before you build, tell me anything you can’t map one to one, and ask me about anything I haven’t said.

Wrap it in a tool. The tool starts the flow with the non-blocking Call Flow API and tells the agent the work is underway:

import os

import requests
from langchain.tools import tool

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


@tool
def send_quote(customer_email: str, customer_name: str, line_items: list[dict]) -> str:
    """Create a QuickBooks estimate for the customer, email it to them, and tell
    the account owner. line_items is a list of {"name", "quantity", "price"}."""
    resp = requests.post(
        SEND_QUOTE_URL,
        headers={"FR-Access-Key": os.environ["FLOWRUNNER_KEY"]},
        json={
            "customerEmail": customer_email,
            "customerName": customer_name,
            "lineItems": line_items,
            "callbackUrl": os.environ["QUOTE_CALLBACK_URL"],
        },
        timeout=30,
    )
    resp.raise_for_status()
    return f"Quote started as FlowRunner run {resp.json()['executionId']}."

Add send_quote to the agent’s tools like any other. When the flow finishes, it POSTs the estimate number to QUOTE_CALLBACK_URL, so your app can store it or tell the user.

For a tool the agent needs an answer from in the same turn, a stock check or a customer lookup, call Call Flow (Blocking) instead: same URL ending in /activate-blocking, with waitResponseTimeoutSeconds up to 300. The response body is whatever the flow’s Return Result block returns. Each run is one execution either way, however many steps the flow takes.

Send middleware approvals to a person

HumanInTheLoopMiddleware pauses the agent before a tool call you list in interrupt_on, and waits for a decision. Getting that request in front of the right person, and their answer back, is what the flow does.

Build the approval flow:

Build a flow called “Refund approval” that starts from the Call Flow API. Its Initial Data is threadId, tool, args (an object with customer, orderId, amount, and reason), description, and callbackUrl. Send the support lead a Slack direct message with the customer, order, amount, and reason, and Approve and Reject buttons, then wait for the click. If there’s no answer in 4 hours, post the same request in #support-leads. When someone answers, POST threadId, decision (approve or reject), any note, and this run’s Execution ID to callbackUrl with an X-Callback-Secret header whose value I’ll paste into the block myself. Retry that POST on 5XX and network errors. Ask me about anything I haven’t said.

Hand off the interrupt. When the agent stops at issue_refund, the app sends the pending call to the flow:

import os

import requests
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=os.environ["AGENT_MODEL"],
    tools=[lookup_order, issue_refund],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={"issue_refund": {"allowed_decisions": ["approve", "reject"]}}
        )
    ],
    checkpointer=InMemorySaver(),  # use a persistent checkpointer in production
)

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


def handle_ticket(ticket_id: str, message: str):
    config = {"configurable": {"thread_id": ticket_id}}
    result = agent.invoke({"messages": [{"role": "user", "content": message}]}, config)
    if "__interrupt__" in result:
        action = result["__interrupt__"][0].value["action_requests"][0]
        requests.post(
            APPROVAL_URL,
            headers={"FR-Access-Key": os.environ["FLOWRUNNER_KEY"]},
            json={
                "threadId": ticket_id,
                "tool": action["name"],
                "args": action["args"],
                "description": action["description"],
                "callbackUrl": os.environ["APPROVAL_CALLBACK_URL"],
            },
            timeout=30,
        ).raise_for_status()
    return result

Resume on the answer. The flow POSTs the decision to APPROVAL_CALLBACK_URL, and the endpoint turns it into the middleware’s decision format:

import hmac
import os

from fastapi import FastAPI, Header, HTTPException
from langgraph.types import Command

app = FastAPI()


@app.post("/approvals/answer")
def approval_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)
    if body["decision"] == "approve":
        decision = {"type": "approve"}
    else:
        decision = {"type": "reject", "message": body.get("note") or "Rejected by the reviewer."}
    config = {"configurable": {"thread_id": body["threadId"]}}
    agent.invoke(Command(resume={"decisions": [decision]}), config)
    return {"ok": True}

A few details:

• Use a persistent checkpointer. The docs say you must configure one to keep state across interrupts, and InMemorySaver is for testing. The answer may come hours or days later, after a deploy.
• Several reviewed calls at once. If the agent can request more than one interrupted tool call in a turn, send them all to the flow and return one decision per call, in the same order as action_requests.
• Letting the reviewer edit. To allow the edit decision, the reviewer needs fields to change. A FlowRunner Form can show the proposed arguments as editable fields, with a public link that needs no account. FlowRunner Prompt to Flow can’t create Forms yet, so it tells you what to add in the flow editor.
• The wait. The run waits up to 30 days on Free, Starter, and Growth, and up to a year on Professional and above, and the request, the reminder, and the answer are on its run record.

Let FlowRunner call your agent

The third direction: something in FlowRunner needs your agent’s judgment.

From a flow, over HTTP. Put the agent behind an endpoint:

@app.post("/agents/triage")
def triage(body: dict):
    result = triage_agent.invoke({"messages": [{"role": "user", "content": body["text"]}]})
    return {"reply": result["messages"][-1].content}

Then build the flow around it:

Build a flow that starts on a new Intercom conversation. POST the conversation text to https://agents.example.com/agents/triage with an Authorization header I’ll fill in with my bearer token in the flow editor, retrying on 5XX. Reply in the conversation with the reply field. Then tag the conversation “ai-triaged” and post a link to it in #support. Ask me about anything I haven’t said.

From a FlowRunner AI agent, over MCP. When a FlowRunner AI Agent block should decide for itself when to ask your agent, expose your agent as an MCP server with the MCP Python SDK:

from mcp.server import MCPServer

mcp = MCPServer("contract-review")


@mcp.tool()
def review_contract(contract_text: str) -> str:
    """Review a contract against our playbook and list the clauses that need a lawyer."""
    result = review_agent.invoke({"messages": [{"role": "user", "content": contract_text}]})
    return result["messages"][-1].content


if __name__ == "__main__":
    mcp.run(transport="streamable-http")

Serve it at a public HTTPS address behind a gateway that checks a token. In FlowRunner, register it under Agent tools & Knowledge > MCP Servers with the URL and the token header, then attach it to an AI Agent block under Manage Capabilities > MCP Extensions (MCP servers in the docs). Registering the server is a console step, because FlowRunner Prompt to Flow can’t do it yet. If your agent runs on LangSmith Deployment, its /mcp endpoint is the other route to the same result.

Optional: turn agent code into a flow

If a LangChain agent is mostly prompt, a few API tools, and an approval, you can also have it on the FlowRunner canvas. FlowRunner Prompt to Flow can read the agent code and build the equivalent flow, with an AI Agent block running on your own model key. Paste the code and ask:

Read this LangChain agent and build the equivalent FlowRunner flow: an AI Agent block with the same instructions, a tool for each of its tools using FlowRunner integrations where one exists, and a human approval wherever the code uses HumanInTheLoopMiddleware. Tell me which tools have no FlowRunner equivalent and should stay as calls to my service, and ask me about anything the code doesn’t say.

It’s an option for agents that are mostly glue. Agents with custom retrievers, your own models, or logic you’ve tuned belong in your code, and the patterns above connect them where they are.