1034206870
- LangflowClient (list/export, x-api-key auth, error mapping) - import_from_langflow: export -> versioned definition -> LFX deploy - POST /api/flows/import, GET /api/flows/langflow/available - scripts/deploy_flow.py CLI - 25 tests green (Langflow faked via dependency override + respx client tests; respx passthrough proven unreliable for LFX traffic)
54 lines
1.6 KiB
Python
54 lines
1.6 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 argparse
|
|
import json
|
|
import sys
|
|
import urllib.request
|
|
|
|
|
|
def post(api: str, path: str, body: dict) -> dict:
|
|
request = urllib.request.Request(
|
|
api.rstrip("/") + path,
|
|
data=json.dumps(body).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
try:
|
|
with urllib.request.urlopen(request) as response:
|
|
return {"status": response.status, "body": json.loads(response.read())}
|
|
except urllib.error.HTTPError as exc:
|
|
return {"status": exc.code, "body": exc.read().decode()[:2000]}
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--api", default="http://localhost:8000")
|
|
parser.add_argument("--langflow-id", required=True)
|
|
parser.add_argument("--name", default=None)
|
|
parser.add_argument("--author", default="")
|
|
parser.add_argument("--no-deploy", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
result = post(
|
|
args.api,
|
|
"/api/flows/import",
|
|
{
|
|
"langflow_flow_id": args.langflow_id,
|
|
"name": args.name,
|
|
"author": args.author,
|
|
"deploy": not args.no_deploy,
|
|
},
|
|
)
|
|
print(json.dumps(result, indent=2, ensure_ascii=False))
|
|
return 0 if result["status"] in (200, 201) else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|