Skip to main content
Version: 1.10.x

Flow management endpoints

Use the /flows endpoint to create, read, update, and delete flows.

If you want to use the Langflow API to run a flow, see Flow trigger endpoints.

Create flow​

Creates a new flow.

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/"

headers = {
    "accept": "application/json",
    "Content-Type": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

payload = {
    "name": "string2",
    "description": "string",
    "icon": "string",
    "icon_bg_color": "#FF0000",
    "gradient": "string",
    "data": {},
    "is_component": False,
    "updated_at": "2024-12-30T15:48:01.519Z",
    "webhook": False,
    "endpoint_name": "string",
    "tags": ["string"],
}

response = requests.request("POST", url, headers=headers, json=payload)
response.raise_for_status()

print(response.text)
Result
{
  "name": "string2",
  "description": "string",
  "icon": "string",
  "icon_bg_color": "#FF0000",
  "gradient": "string",
  "data": {},
  "is_component": false,
  "updated_at": "2025-02-04T21:07:36+00:00",
  "webhook": false,
  "endpoint_name": "string",
  "tags": ["string"],
  "locked": false,
  "id": "e8d81c37-714b-49ae-ba82-e61141f020ee",
  "user_id": "f58396d4-a387-4bb8-b749-f40825c3d9f3",
  "project_id": "1415de42-8f01-4f36-bf34-539f23e47466"
}

Create flows​

Creates multiple new flows, returning an array of flow objects.

import os
import uuid

import requests

base = os.environ.get("LANGFLOW_URL", "")
api_key = os.environ.get("LANGFLOW_API_KEY", "")
folder_id = (os.environ.get("PROJECT_ID") or os.environ.get("FOLDER_ID") or "").strip()

headers = {"accept": "application/json", "Content-Type": "application/json", "x-api-key": api_key}


def _flow_doc(suffix: str) -> dict:
    doc = {
        "name": f"batch-flow-{uuid.uuid4().hex[:8]}",
        "description": f"Docs batch example {suffix}",
        "data": {"nodes": [], "edges": []},
    }
    if folder_id:
        doc["folder_id"] = folder_id
    return doc


payload = {"flows": [_flow_doc("A"), _flow_doc("B")]}

response = requests.post(f"{base}/api/v1/flows/batch/", headers=headers, json=payload, timeout=30)
response.raise_for_status()
print(response.text)

Read flow​

Retrieves a specific flow by its ID.

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/{os.getenv('FLOW_ID', '')}"

headers = {
    "accept": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

response = requests.request("GET", url, headers=headers)
response.raise_for_status()

print(response.text)
Result
{
  "name": "Basic Prompting",
  "description": "Perform basic prompting with an OpenAI model.",
  "icon": "Braces",
  "icon_bg_color": null,
  "gradient": "2",
  "data": {
    "nodes": [
     ...
    ]
  }
}

Read flows​

Returns a JSON object containing a list of flows.

Retrieve all flows with pagination:

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/?remove_example_flows=false&components_only=false&get_all=true&header_flows=false&page=1&size=50"

headers = {
    "accept": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

response = requests.request("GET", url, headers=headers)
response.raise_for_status()

print(response.text)

To retrieve flows from a specific project, use the project_id query parameter:

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/?remove_example_flows=true&components_only=false&get_all=false&project_id={os.getenv('PROJECT_ID', '')}&header_flows=false&page=1&size=1"

headers = {
    "accept": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

response = requests.request("GET", url, headers=headers)
response.raise_for_status()

print(response.text)

Read sample flows​

Retrieves a list of sample flows:

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/basic_examples/"

headers = {
    "accept": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

response = requests.request("GET", url, headers=headers)
response.raise_for_status()

print(response.text)

Update flow​

Updates an existing flow by its ID.

This example changes the value for endpoint_name from a random UUID to my_new_endpoint_name.

import os
import uuid

import requests

base = os.environ.get("LANGFLOW_URL") or os.environ.get("LANGFLOW_SERVER_URL", "")
flow_id = os.environ.get("FLOW_ID", "")
api_key = os.environ.get("LANGFLOW_API_KEY", "")

headers = {"accept": "application/json", "Content-Type": "application/json", "x-api-key": api_key}

payload = {
    "name": f"docs-example-updated-flow-{uuid.uuid4().hex[:8]}",
    "description": "Updated via API docs Python example",
    "locked": False,
}

response = requests.patch(f"{base}/api/v1/flows/{flow_id}", headers=headers, json=payload, timeout=30)
response.raise_for_status()
print(response.text)
Result
{
  "name": "string",
  "description": "string",
  "icon": "Braces",
  "icon_bg_color": null,
  "gradient": "2",
  "data": {},
  "is_component": false,
  "updated_at": "2024-12-30T18:30:22+00:00",
  "webhook": false,
  "endpoint_name": "my_new_endpoint_name",
  "tags": null,
  "locked": true,
  "id": "01ce083d-748b-4b8d-97b6-33adbb6a528a",
  "user_id": "f58396d4-a387-4bb8-b749-f40825c3d9f3",
  "project_id": "3fa85f64-5717-4562-b3fc-2c963f66afa6"
}

Delete flow​

Deletes a specific flow by its ID.

import os

import requests

base = os.environ.get("LANGFLOW_URL", "")
api_key = os.environ.get("LANGFLOW_API_KEY", "")

headers = {"accept": "application/json", "Content-Type": "application/json", "x-api-key": api_key}

create = requests.post(
    f"{base}/api/v1/flows/",
    headers=headers,
    json={
        "name": "docs-example-delete-me",
        "description": "Temporary flow for delete-flow example",
        "data": {"nodes": [], "edges": []},
    },
    timeout=30,
)
create.raise_for_status()
flow_id = create.json()["id"]

delete = requests.delete(f"{base}/api/v1/flows/{flow_id}", headers=headers, timeout=30)
delete.raise_for_status()
print(delete.text)
Result
{
  "message": "Flow deleted successfully"
}

Export flows​

Exports specified flows to a ZIP file.

This endpoint downloads a ZIP file containing Langflow JSON files for each flow ID listed in the request body.

import os

import requests

base = os.environ.get("LANGFLOW_URL", "")
api_key = os.environ.get("LANGFLOW_API_KEY", "")
flow_id = os.environ.get("FLOW_ID", "")
folder_id = os.environ.get("PROJECT_ID") or os.environ.get("FOLDER_ID", "")

headers = {
    "accept": "application/json",
    "Content-Type": "application/json",
    "x-api-key": api_key,
}

# Export needs at least two flows to return a ZIP; a single id returns JSON.
extra = requests.post(
    f"{base}/api/v1/flows/",
    headers=headers,
    json={
        "name": "docs-export-temp-flow",
        "description": "Temporary second flow for export example",
        "data": {"nodes": [], "edges": []},
        **({"folder_id": folder_id} if folder_id else {}),
    },
    timeout=30,
)
extra.raise_for_status()
extra_id = extra.json()["id"]

payload = [flow_id, extra_id]

response = requests.post(f"{base}/api/v1/flows/download/", headers=headers, json=payload, timeout=60)
response.raise_for_status()

with open("langflow-flows.zip", "wb") as f:
    f.write(response.content)
print("Saved response to langflow-flows.zip")

requests.delete(f"{base}/api/v1/flows/{extra_id}", headers=headers, timeout=30)
Result
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 76437    0 76353  100    84  4516k   5088 --:--:-- --:--:-- --:--:-- 4665k

Import flows​

Imports flows by uploading a Langflow-compatible JSON file.

To specify a target project for the flow, include the query parameter folder_id. The target folder_id must already exist before uploading a flow. Call the /api/v1/projects/ endpoint for a list of available folders and projects.

This example uploads a local file named agent-with-astra-db-tool.json to a folder specified by a FOLDER_ID variable:

import os

import requests

url = f"{os.getenv('LANGFLOW_URL', '')}/api/v1/flows/upload/?folder_id={os.getenv('FOLDER_ID', '')}"

headers = {
    "accept": "application/json",
    "x-api-key": f"{os.getenv('LANGFLOW_API_KEY', '')}",
}

files = {
    "file": open(os.getenv("FLOW_IMPORT_FILE", "docs/docs/API-Reference/fixtures/flow-import.json"), "rb"),
}

response = requests.request("POST", url, headers=headers, files=files)
response.raise_for_status()

print(response.text)

for _f in files.values():
    if hasattr(_f, "close"):
        _f.close()
Result
[
  {
    "name": "agent-with-astra-db-tool",
    "description": "",
    "icon": null,
    "icon_bg_color": null,
    "gradient": null,
    "data": {}
  ...
  }
]

Was this page helpful?

Support
Search