LangGraph 系列 4：使用 SDK 和 REST 调用 LangGraph Agent API

上一篇已经使用 langgraph dev 启动天气 Agent，并在 Studio 中看到了模型调用工具的完整过程。但 Studio 主要用于开发和调试，真正开发 Web 页面、后端服务或命令行工具时，程序还需要通过 API 调用 Agent。

本篇继续使用已经创建好的 weather_agent，分别通过 Python 异步 SDK、Python 同步 SDK、JavaScript SDK 和 curl 调用同一个 Agent Server。重点不是再写一个 Agent，而是理解客户端如何发起 Run，以及服务端怎样把执行过程持续返回给客户端。

1. 先分清两个本地接口

本例同时运行两个 HTTP 服务，它们不能混用：

服务	地址	调用者	作用
Qwen3 MLX Server	http://127.0.0.1:18080/v1	天气 Agent 内部的 ChatOpenAI	负责模型推理
LangGraph Agent Server	http://127.0.0.1:2024	Python、JavaScript 和 curl 客户端	加载 Graph、创建 Run、执行工具并返回事件流

客户端应该访问 2024 端口，而不是直接访问 18080。如果直接请求 18080，得到的只是模型能力，不会经过 LangGraph，也不会执行天气工具。

下图先从服务边界说明四类客户端实际连接到了哪里。SDK 只是对 Agent API 的封装，curl 则直接发送 HTTP 请求；它们都不会越过 Agent Server 直接调用天气工具。

一次完整请求的过程如下：

1. 客户端向 Agent Server 提交用户消息。
2. Agent Server 根据 assistant_id 找到 weather_agent。
3. Agent 调用 Qwen3，判断是否需要使用天气工具。
4. Agent Server 执行 get_weather()，再把工具结果交给 Qwen3。
5. Agent Server 通过 SSE 持续返回工具调用、工具结果和最终回答。

这里需要特别注意：Qwen3 生成的是“调用 get_weather”的结构化请求，真正执行 Python 函数的是 Agent Server 中的工具节点。模型收到 ToolMessage 后，才会继续生成面向用户的最终回答。

2. 启动 Qwen3 和 Agent Server

四个客户端都依赖这两个服务。运行客户端之前，应先分别检查两个服务是否可用。

2.1 启动 Qwen3

打开第一个终端，进入 llm_learning 根目录，复用 LangGraph 系列 3 创建的模型服务环境：

cd source/_posts/llm_learning
source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model model \
  --host 127.0.0.1 \
  --port 18080 \
  --chat-template-args '{"enable_thinking": false}'

另开终端检查模型接口：

curl --noproxy '*' \
  http://127.0.0.1:18080/v1/models

只要返回结果中的 object 是 list，并且 data 中存在模型，说明模型服务已经就绪。

2.2 启动 Agent Server

打开第二个终端：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate
cd langgraph/p02_local_agent_project

langgraph dev \
  --host 127.0.0.1 \
  --port 2024 \
  --no-browser

检查健康接口：

curl --noproxy '*' \
  http://127.0.0.1:2024/ok

真实返回结果：

{"ok": true}

如果两个健康检查没有通过，应先解决服务问题，再运行客户端。这样可以避免把模型未启动、Graph 加载失败等问题误认为 SDK 错误。

3. 安装 Python LangGraph SDK

本篇继续使用 LangGraph 系列 2 创建的 .venv_langgraph，不再创建新的 Python 环境：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate

python -m pip install \
  -r langgraph/p04_agent_api_client/requirements.txt

requirements.txt 只有一个直接依赖：

langgraph-sdk==0.3.12

检查安装结果：

python -m pip show langgraph-sdk
python -m pip check

langgraph-sdk 是 Agent Server 的 Python 客户端。它封装了 Assistants、Threads 和 Runs 等 HTTP 接口，本篇只使用 runs.stream() 创建一次运行并接收流式结果。详细接口可以查阅 LangGraph Python SDK API Reference。

4. 调用 Run 时的五个关键参数

四种客户端写法不同，但最终都需要提供相同的关键信息。

4.1 url

http://127.0.0.1:2024

这是 Agent Server 的地址。不要写成 Qwen3 的 18080/v1。

4.2 thread_id

本篇传入 None 或 null：

None

这会创建一次无状态运行，不绑定可复用的 Thread。它适合先验证 API 调用链，但后续请求不能依靠这个参数恢复本次会话。Agent Server 将这种调用归入 Stateless Runs；需要保留多轮状态时，应先创建 Thread，再把 thread_id 传给 Run。相关概念可以参考 Runs 官方说明。

4.3 assistant_id

weather_agent

这个名称来自 LangGraph 系列 2 的 langgraph.json：

{
  "graphs": {
    "weather_agent": "./src/weather_agent/graph.py:graph"
  }
}

课程或其他项目中可能使用 agent，但本项目注册的名称是 weather_agent。名称不一致时，Agent Server 无法找到对应 Graph。

4.4 input

输入需要符合 Graph 的状态结构。天气 Agent 接收消息列表：

input={
    "messages": [
        {"role": "human", "content": "北京天气怎么样？"},
    ]
}

4.5 stream_mode

本篇使用：

messages-tuple

它会以消息块为单位返回模型和工具产生的消息。客户端既能看到工具调用和工具结果，也能逐段接收最终回答。

5. 使用 Python 异步 SDK

异步客户端使用 get_client()。调用 runs.stream() 后，async for 会在事件到达时逐条处理，不需要等 Agent 完全执行结束才得到结果。

langgraph/p04_agent_api_client/01_call_agent_async.py 的完整代码如下：

"""使用 Python 异步 SDK 调用天气 Agent，并观察关键执行事件。"""

import asyncio

from langgraph_sdk import get_client


async def main() -> None:
    """连接本地 Agent Server，并以流式方式读取执行结果。"""

    # 这里连接的是 LangGraph Agent Server，不是 18080 端口的模型服务。
    client = get_client(url="http://127.0.0.1:2024")
    tool_finished = False
    answer_started = False

    async for chunk in client.runs.stream(
        None,  # 不指定可复用的 thread_id，本次使用临时运行。
        "weather_agent",  # 对应 langgraph.json 中注册的 Graph 名称。
        input={
            "messages": [
                {"role": "human", "content": "北京天气怎么样？"},
            ]
        },
        stream_mode="messages-tuple",
    ):
        if chunk.event == "metadata":
            # metadata 中包含动态 run_id，这里只说明事件已经到达。
            print("运行事件：Agent Run 已创建")
            continue

        if chunk.event != "messages":
            continue

        # messages-tuple 的第一项是消息块，第二项是执行元数据。
        message, metadata = chunk.data
        message_type = message.get("type")

        if message.get("tool_calls"):
            for tool_call in message["tool_calls"]:
                print(f"工具调用：{tool_call['name']}({tool_call['args']})")
            continue

        if message_type == "tool":
            tool_finished = True
            print(f"工具结果：{message['content']}")
            continue

        content = message.get("content", "")
        if message_type == "AIMessageChunk" and tool_finished and content:
            if not answer_started:
                print("最终回答：", end="", flush=True)
                answer_started = True
            print(content, end="", flush=True)

    print()


if __name__ == "__main__":
    asyncio.run(main())

运行：

python \
  langgraph/p04_agent_api_client/01_call_agent_async.py

一次真实运行结果如下。动态 Run ID 和消息元数据没有打印：

运行事件：Agent Run 已创建
工具调用：get_weather({'city': '北京'})
工具结果：北京今天的天气晴朗，气温是 28 摄氏度。
最终回答：北京今天的天气晴朗，气温是 28 摄氏度。

这段输出证明客户端接收到的并不只有最终答案。工具调用和工具结果同样属于 Agent 的执行过程。

天气数据来自本地模拟函数，不是实时天气接口。模型每次生成的措辞可能略有不同，但工具名、工具结果和事件结构应保持一致。

6. messages-tuple 中有什么

/runs/stream 在 HTTP 层返回 SSE 文本。Python SDK 收到这些文本后，会把每组 event: 和 data: 解析成 chunk 对象，因此 SDK 返回的每个 chunk 至少包含两个重要字段：

chunk.event
chunk.data

当事件类型为 messages 时，chunk.data 才是一个二元结构：

message, metadata = chunk.data

• message：当前消息或消息块，例如工具调用、ToolMessage、AIMessageChunk。
• metadata：当前消息来自哪个节点、处于哪一步等执行信息。

图的上半部分是原始 SSE 传输格式，中间是 SDK 解析后的对象，下半部分是本例重新实测得到的事件顺序。metadata 控制事件不使用 [message, metadata] 结构；只有 messages 事件的 data 才能拆成这两个部分。

本例读取了消息附带的 metadata，但没有完整打印。原因不是它没有作用，而是其中包含 Run ID、节点执行信息等动态字段。教学输出只保留判断执行过程需要的稳定内容，更容易看清工具调用链。

事件顺序可以概括为：

metadata
→ AIMessageChunk，包含 get_weather 工具调用
→ ToolMessage，包含天气工具结果
→ AIMessageChunk，逐段生成最终回答

7. 使用 Python 同步 SDK

同步客户端使用 get_sync_client()。它不需要 asyncio 和 async for，更适合简单脚本或暂时没有异步运行环境的程序。

langgraph/p04_agent_api_client/02_call_agent_sync.py 的完整代码如下：

"""使用 Python 同步 SDK 调用天气 Agent，并流式输出最终回答。"""

from langgraph_sdk import get_sync_client


# 同步客户端会阻塞当前线程，但服务端返回的内容仍然可以流式读取。
client = get_sync_client(url="http://127.0.0.1:2024")
tool_finished = False

print("LangGraph Agent 回答：", end="", flush=True)

for chunk in client.runs.stream(
    None,  # 本例不保存可复用的会话线程。
    "weather_agent",
    input={
        "messages": [
            {"role": "human", "content": "北京天气怎么样？"},
        ]
    },
    stream_mode="messages-tuple",
):
    if chunk.event != "messages":
        continue

    # messages-tuple 返回 [message_chunk, metadata]。
    message, metadata = chunk.data

    if message.get("type") == "tool":
        tool_finished = True
        continue

    content = message.get("content", "")
    if message.get("type") == "AIMessageChunk" and tool_finished and content:
        print(content, end="", flush=True)

print()

运行：

python \
  langgraph/p04_agent_api_client/02_call_agent_sync.py

一次真实运行结果：

LangGraph Agent 回答：北京今天的天气晴朗，气温是 28 摄氏度。

这个脚本仍然使用流式接口，只是过滤掉了工具调用等中间消息，仅把工具执行后的最终 AI 消息逐段写到终端。

8. 同步、异步和流式不是一回事

这三个概念经常被混在一起：

• 同步：调用期间当前线程等待结果，代码使用普通 for。
• 异步：等待网络数据时可以让出执行权，代码使用 async for。
• 流式：服务端不是等全部内容完成后一次返回，而是连续发送事件或消息块。

因此，同步客户端同样可以接收流式结果；异步客户端也可以调用非流式接口。同步或异步描述客户端怎样等待，流式描述结果怎样返回。

9. 使用 JavaScript SDK

Node.js 服务可以使用 @langchain/langgraph-sdk 调用同一个 Agent Server。本例使用 process.stdout.write() 输出终端内容，因此是一段 Node.js 脚本，不是可以原样放进浏览器的代码。浏览器接入时还要处理 Agent Server 的跨域配置、认证方式和页面状态管理。

进入示例目录安装固定依赖：

cd source/_posts/llm_learning/\
langgraph/p04_agent_api_client

npm ci

package.json 固定 SDK 版本，并将会随时间变化的传递依赖锁定到本文环境实际使用的版本：

{
  "name": "langgraph-agent-api-client",
  "version": "1.0.0",
  "private": true,
  "type": "module",
  "dependencies": {
    "@langchain/langgraph-sdk": "1.8.2"
  },
  "overrides": {
    "is-network-error": "1.3.1",
    "p-queue": "9.1.1",
    "uuid": "13.0.0"
  }
}

langgraph/p04_agent_api_client/03_call_agent_javascript.mjs 的完整代码如下：

/**
 * 使用 JavaScript SDK 调用本地天气 Agent，并流式输出最终回答。
 */

import { Client } from "@langchain/langgraph-sdk";

// apiUrl 指向 LangGraph Agent Server，而不是 Qwen3 模型端口。
const client = new Client({ apiUrl: "http://127.0.0.1:2024" });

const stream = client.runs.stream(null, "weather_agent", {
  input: {
    messages: [{ role: "human", content: "北京天气怎么样？" }],
  },
  streamMode: "messages-tuple",
});

let toolFinished = false;
process.stdout.write("JavaScript SDK 回答：");

for await (const chunk of stream) {
  if (chunk.event !== "messages") {
    continue;
  }

  // messages-tuple 的第一项是消息块，第二项是执行元数据。
  const [message, metadata] = chunk.data;

  if (message.type === "tool") {
    toolFinished = true;
    continue;
  }

  if (message.type === "AIMessageChunk" && toolFinished && message.content) {
    process.stdout.write(message.content);
  }
}

process.stdout.write("\n");

运行：

node 03_call_agent_javascript.mjs

一次真实运行结果：

JavaScript SDK 回答：北京今天的天气晴朗，气温是 28 摄氏度。

Python 和 JavaScript 客户端并没有维护两套 Agent。它们使用不同语言封装相同的 Agent API，最终请求的都是 weather_agent。

10. 使用 curl 调用 REST API

SDK 最终仍然通过 HTTP 请求 Agent Server。使用 curl 可以直接看到请求地址、JSON 请求体和 SSE 响应方式，有助于理解 SDK 隐藏了哪些细节。

langgraph/p04_agent_api_client/04_call_agent_rest.sh 的完整代码如下：

#!/usr/bin/env bash

# 使用原生 REST API 调用天气 Agent，并汇总收到的 SSE 事件。

set -euo pipefail

curl --noproxy '*' -sS -N \
  --request POST \
  --url http://127.0.0.1:2024/runs/stream \
  --header 'Content-Type: application/json' \
  --data '{
    "assistant_id": "weather_agent",
    "input": {
      "messages": [
        {"role": "human", "content": "北京天气怎么样？"}
      ]
    },
    "stream_mode": "messages-tuple"
  }' |
  awk '
    /^event: metadata/ { metadata_count++ }
    /^event: messages/ { message_count++ }
    END {
      print "metadata 事件数量：" metadata_count
      print "messages 事件数量：" message_count
    }
  '

运行：

bash langgraph/p04_agent_api_client/04_call_agent_rest.sh

一次真实运行结果：

metadata 事件数量：1
messages 事件数量：20

消息事件数量会受到模型分词和生成结果影响，不保证每次都等于 20。这次输出只记录本次真实执行结果。

脚本中的 awk 只统计事件，没有显示动态 Run ID 和整段消息。如果希望观察原始响应，可以暂时移除管道符之后的 awk。原始 SSE 响应由多组下面这样的字段组成：

event: metadata
data: {本次运行的元数据}

event: messages
data: [消息块, 执行元数据]

这里的大括号内容是结构说明，不是伪造的运行值。真实 data 中会包含动态 UUID 和节点元数据，不适合写成固定结果。

11. SSE 中的 event 和 data

/runs/stream 返回 text/event-stream，也就是 Server-Sent Events，简称 SSE。服务端可以在一条 HTTP 连接上连续向客户端发送事件。Agent Server 将 Graph 的执行过程组织为 Run，并通过流式端点返回过程事件；Assistants、Threads、Runs 等接口边界可以参考 Agent Server 官方说明。

每组事件主要包含：

• event:：事件名称，例如 metadata 或 messages。
• data:：该事件携带的 JSON 数据。
• 空行：表示一组 SSE 事件结束。

Python 和 JavaScript SDK 已经把原始文本解析成对象，所以代码中可以直接读取：

chunk.event
chunk.data

curl 没有做这层封装，因此看到的是原始 event: 和 data: 文本。两者收到的是同一条执行事件流，只是表示方式不同。

12. 如何识别工具调用和最终回答

messages-tuple 会返回多个消息块，不能把第一段 AI 内容直接当成最终回答。本例按照下面的顺序判断：

1. 消息包含 tool_calls：模型请求调用工具。
2. 消息类型是 tool：工具已经执行并返回结果。
3. 工具完成后出现非空 AIMessageChunk：这是最终回答的内容块。

同步和 JavaScript 示例使用 toolFinished 标记，就是为了跳过工具执行前的模型消息，只输出工具完成后的回答。

这种判断针对当前天气 Agent 的简单执行链。更复杂的 Agent 可能多次调用工具，或者包含多个节点，客户端需要根据产品需求决定展示哪些事件。

13. 常见问题

13.1 连接错端口

客户端 SDK 应连接：

http://127.0.0.1:2024

18080/v1 是模型接口，不提供 /runs/stream。

13.2 Agent Server 正常，但调用时报 502

先检查模型服务：

curl --noproxy '*' \
  http://127.0.0.1:18080/v1/models

Agent Server 可以在 Qwen3 没有启动时打开健康接口，但运行到模型节点时会失败。

13.3 assistant_id 写错

本项目使用：

weather_agent

它必须与 langgraph.json 的 graphs 键一致。不要直接照抄其他项目中的 agent。

13.4 stream_mode 拼写错误

Python 使用下划线参数名：

stream_mode="messages-tuple"

JavaScript 使用驼峰参数名：

streamMode: "messages-tuple"

REST JSON 使用：

"stream_mode": "messages-tuple"

13.5 把同步误认为非流式

get_sync_client() 只是同步等待，client.runs.stream() 仍然会逐段返回结果。是否流式由调用的接口和 stream_mode 决定。

13.6 VPN 或系统代理影响本机请求

curl 使用了：

--noproxy '*'

如果本机请求仍被代理软件接管，应先把 127.0.0.1 和 localhost 加入代理绕过列表，再检查两个健康接口。

13.7 输出中出现大量 UUID 和元数据

直接 print(chunk) 会把 Run ID、节点信息和完整消息对象全部打印出来。排查底层问题时这些字段很有用，但学习调用流程时应按 chunk.event 和消息类型筛选，保留真正需要的内容。

14. 总结

Agent Server 将本地 Graph 变成了统一的 HTTP API。Python、JavaScript 和 curl 虽然写法不同，但都在做同一件事：向 weather_agent 提交消息，创建一次 Run，再通过 SSE 接收执行事件。

Python SDK 适合 Python 后端和自动化脚本，JavaScript SDK 适合 Node.js 或前端相关项目，curl 适合验证接口和排查原始请求。实际开发时通常选择一种 SDK，出现问题时再用 curl 检查 HTTP 和 SSE 层。

本篇仍然使用无状态运行，没有展开可复用 Thread、检查点和持久化。下一步理解这些能力时，需要继续区分 Run、Thread 和 Graph 状态，而不是把所有历史记录都放在客户端。

调用方式	客户端形式	是否可以流式	本例输出	适合场景
Python 异步 SDK	get_client() + async for	是	工具调用、工具结果、最终回答	异步 Web 服务、并发任务
Python 同步 SDK	get_sync_client() + for	是	最终回答	命令行脚本、同步程序
JavaScript SDK	Client + for await	是	最终回答	Node.js、Web 客户端相关项目
curl REST	POST /runs/stream	是，原始 SSE	event: 和 data:	接口验证、协议排查