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 /path/to/llm-learning
source .venv_tool_server/bin/activate

"$VIRTUAL_ENV/bin/python" -m mlx_lm server \
  --model Qwen3-14B-AWQ-4bit-MLX \
  --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 /path/to/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 /path/to/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，并以流式方式读取执行结果。"""

    # 第 1 步：创建异步 SDK 客户端。
    # 这里连接的是 LangGraph Agent Server，不是 18080 端口的模型服务。
    client = get_client(url="http://127.0.0.1:2024")
    tool_finished = False
    answer_started = False

    # 第 2 步：发送问题，并逐块读取服务端返回的事件。
    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":
            # type=tool 表示工具已经执行完成。
            tool_finished = True
            print(f"工具结果：{message['content']}")
            continue

        content = message.get("content", "")
        if message_type == "AIMessageChunk" and tool_finished and content:
            # 工具完成后出现的 AI 文本块，才属于最终自然语言回答。
            if not answer_started:
                print("最终回答：", end="", flush=True)
                answer_started = True
            print(content, end="", flush=True)

    print()


# 只有直接运行当前文件时才启动异步 main()。
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. 使用 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

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

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

# 第 2 步：发送问题，并用普通 for 循环读取流式事件。
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":
        # 工具完成后，后续 AIMessageChunk 才是最终回答。
        tool_finished = True
        continue

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

# 流式输出使用 end=""，最后补一个换行。
print()

运行：

python langgraph/p04_agent_api_client/02_call_agent_sync.py

一次真实运行结果：

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

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

7. 使用 JavaScript SDK

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

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

cd /path/to/llm-learning/\
langgraph/p04_agent_api_client

npm ci

package.json 声明 SDK 依赖，package-lock.json 记录完整的依赖解析结果：

{
  "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。

8. 使用 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 和节点元数据，不适合写成固定结果。

9. 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：当前消息来自哪个节点、处于哪一步等执行信息。本例没有读取它，因此源码按 Python 惯例命名为 _metadata。

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

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

事件顺序可以概括为：

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

10. 同步、异步和流式

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

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

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

11. 总结

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:	接口验证、协议排查