52 lines
1.5 KiB
Python
52 lines
1.5 KiB
Python
"""Import a flow from Langflow into our backend via the running API.
|
|
|
|
Usage:
|
|
uv run python scripts/deploy_flow.py --api http://localhost:8000 \\
|
|
--langflow-id <uuid> [--name my-flow] [--author jane] [--no-deploy]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import httpx
|
|
import typer
|
|
|
|
|
|
def post(api: str, path: str, body: dict) -> dict:
|
|
try:
|
|
response = httpx.post(api.rstrip("/") + path, json=body, timeout=60.0)
|
|
try:
|
|
payload = response.json()
|
|
except ValueError:
|
|
payload = response.text[:2000]
|
|
return {"status": response.status_code, "body": payload}
|
|
except httpx.HTTPError as exc:
|
|
return {"status": 0, "body": str(exc)[:2000]}
|
|
|
|
|
|
def main(
|
|
langflow_id: str = typer.Option(..., "--langflow-id"),
|
|
api: str = typer.Option("http://localhost:8000", "--api"),
|
|
name: str | None = typer.Option(None, "--name"),
|
|
author: str = typer.Option("", "--author"),
|
|
no_deploy: bool = typer.Option(False, "--no-deploy"),
|
|
) -> None:
|
|
"""Import a flow from Langflow via the running API (post() stays importable)."""
|
|
result = post(
|
|
api,
|
|
"/api/flows/import",
|
|
{
|
|
"langflow_flow_id": langflow_id,
|
|
"name": name,
|
|
"author": author,
|
|
"deploy": not no_deploy,
|
|
},
|
|
)
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
if result["status"] not in (200, 201):
|
|
raise typer.Exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
typer.run(main)
|