用 Langflow Run API 构建安全的流程调用客户端

流程发布后，用 POST /api/v1/run/{flow_id} 调用，头里带 x-api-key。响应是嵌套的 outputs 结构，直接 ["text"] 会取不到。 这篇做 run_client.py（标准库）和 mock：验证 200、401，以及拒绝危险 tweaks 字段。

先看清成品

• mock 返回 echo:食堂几点开门？；extract_text 能取出嵌套字段。
• 401 与拒绝 tweaks 已覆盖；3 项 pytest 通过。
• 未连接真实 Langflow。

合规提醒：用自己的 Langflow 实例与模型密钥；自定义组件里的代码会在服务器上执行，不要加载不可信流程。密钥放环境变量，导出流程前确认不会带出 Key。费用以模型供应商为准，本文不写价格。

运行说明：我没有安装完整 Langflow 服务。真实运行过的是：组件骨架检查、normalize 逻辑、Run API 客户端对 mock 的请求、.env 检查，以及 pytest（2026-10-23，UTC+8）。服务端行为来自官方文档。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

92-langflow-B-src/
├── run_client.py
├── mock_langflow.py
└── test_92b.py

第 1 步：客户端

#!/usr/bin/env python3
"""Call Langflow POST /api/v1/run/{flow_id} with x-api-key (official shape)."""
from __future__ import annotations

import argparse
import json
import os
import sys
import urllib.error
import urllib.request


def run_flow(
    *,
    base_url: str,
    api_key: str,
    flow_id: str,
    input_value: str,
    input_type: str = "chat",
    output_type: str = "chat",
    session_id: str | None = None,
    tweaks: dict | None = None,
) -> dict:
    url = f"{base_url.rstrip('/')}/api/v1/run/{flow_id}"
    payload = {
        "input_value": input_value,
        "input_type": input_type,
        "output_type": output_type,
    }
    if session_id:
        payload["session_id"] = session_id
    if tweaks:
        payload["tweaks"] = tweaks
    req = urllib.request.Request(
        url,
        data=json.dumps(payload).encode(),
        headers={
            "Content-Type": "application/json",
            "x-api-key": api_key,
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            return json.loads(resp.read().decode())
    except urllib.error.HTTPError as e:
        raise SystemExit(f"HTTP {e.code}: {e.read().decode(errors='replace')}") from e


def extract_text(resp: dict) -> str:
    try:
        return resp["outputs"][0]["outputs"][0]["results"]["message"]["text"]
    except (KeyError, IndexError, TypeError):
        return json.dumps(resp, ensure_ascii=False)


def main(argv=None) -> int:
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument("text", nargs="?", default="你好")
    p.add_argument("--base", default=os.environ.get("LANGFLOW_BASE_URL", "http://127.0.0.1:7860"))
    p.add_argument("--key", default=os.environ.get("LANGFLOW_API_KEY", ""))
    p.add_argument("--flow", default=os.environ.get("LANGFLOW_FLOW_ID", ""))
    args = p.parse_args(argv)
    if not args.key or not args.flow:
        print("需要 LANGFLOW_API_KEY 和 LANGFLOW_FLOW_ID", file=sys.stderr)
        return 2
    r = run_flow(base_url=args.base, api_key=args.key, flow_id=args.flow, input_value=args.text)
    print(extract_text(r))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

第 2 步：Mock

#!/usr/bin/env python3
from __future__ import annotations
import json
from http.server import BaseHTTPRequestHandler, HTTPServer
from urllib.parse import urlparse, parse_qs

API_KEY = "lf-test-key"
FLOW_ID = "11111111-2222-3333-4444-555555555555"


class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass

    def do_POST(self):
        u = urlparse(self.path)
        if not u.path.startswith("/api/v1/run/"):
            self.send_error(404)
            return
        flow = u.path.rsplit("/", 1)[-1]
        key = self.headers.get("x-api-key") or parse_qs(u.query).get("x-api-key", [None])[0]
        if key != API_KEY:
            self.send_response(401)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"detail":"Invalid API key"}')
            return
        if flow != FLOW_ID:
            self.send_response(404)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"detail":"Flow not found"}')
            return
        n = int(self.headers.get("Content-Length", 0))
        body = json.loads(self.rfile.read(n) or b"{}")
        tweaks = body.get("tweaks") or {}
        if "refused_field" in json.dumps(tweaks):
            self.send_response(422)
            self.send_header("Content-Type", "application/json")
            self.end_headers()
            self.wfile.write(b'{"detail":"TWEAKS_REFUSED"}')
            return
        text = f"echo:{body.get('input_value','')}"
        resp = {
            "session_id": body.get("session_id") or "sess",
            "outputs": [{"outputs": [{"results": {"message": {"text": text}}}]}],
        }
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.end_headers()
        self.wfile.write(json.dumps(resp).encode())


if __name__ == "__main__":
    import os
    HTTPServer(("127.0.0.1", int(os.environ.get("PORT", "3092"))), H).serve_forever()

第 3 步：测试

import threading
from http.server import HTTPServer
import mock_langflow
from run_client import run_flow, extract_text

def serve():
    httpd = HTTPServer(("127.0.0.1", 0), mock_langflow.H)
    threading.Thread(target=httpd.serve_forever, daemon=True).start()
    return httpd, httpd.server_address[1]

def test_ok():
    httpd, port = serve()
    try:
        r = run_flow(base_url=f"http://127.0.0.1:{port}", api_key=mock_langflow.API_KEY,
                     flow_id=mock_langflow.FLOW_ID, input_value="午餐")
        assert extract_text(r) == "echo:午餐"
    finally:
        httpd.shutdown()

def test_401():
    httpd, port = serve()
    try:
        try:
            run_flow(base_url=f"http://127.0.0.1:{port}", api_key="x",
                     flow_id=mock_langflow.FLOW_ID, input_value="a")
            assert False
        except SystemExit as e:
            assert "401" in str(e)
    finally:
        httpd.shutdown()

def test_tweaks_refused():
    httpd, port = serve()
    try:
        try:
            run_flow(base_url=f"http://127.0.0.1:{port}", api_key=mock_langflow.API_KEY,
                     flow_id=mock_langflow.FLOW_ID, input_value="a",
                     tweaks={"n": {"refused_field": 1}})
            assert False
        except SystemExit as e:
            assert "422" in str(e)
    finally:
        httpd.shutdown()

第 4 步：运行

# 2026-10-23 12:37 UTC+8（mock）
echo:食堂几点开门？
---json---
{
  "session_id": "sess",
  "outputs": [
    {
      "outputs": [
        {
          "results": {
            "message": {
              "text": "echo:食堂几点开门？"
            }
          }
        }
      ]
    }
  ]
}

$ python -m pytest -q
3 passed

怎么检查结果？

• extract_text 得到非 JSON 原文。
• 错误 key → 401。
• 接真服务：设 LANGFLOW_BASE_URL / LANGFLOW_API_KEY / LANGFLOW_FLOW_ID。

调试示例

tweaks 是双刃剑。 官方允许用 tweaks 覆盖节点参数，方便调试；若开放给终端用户，可能被改成指向内网地址或换模型。mock 对名为 refused_field 的 tweak 直接 400，提醒：对外 API 应白名单 tweaks，或干脆不接收。

常见问题

模板规则

• 对外接口白名单 tweaks。
• 密钥放环境变量。
• 先 mock 再接真。

项目资源

嵌套响应不可怕，先写一个 extract_text 固定住。