MCP引入一段简单的LangGraph

发表于 分类于 阅读次数： 本文字数： 837 阅读时长 ≈ 3 分钟

前言

在上一篇文章中，我们讨论了怎么在构建MCP。为了能够实现更复杂的功能，我们尝试将LangGraph引入到MCP中。

一段简单的意图识别

我们就不用大模型引入乱七八糟的机制了。就一个功能，检测输入进来的字符串有多长，奇数个字符就输出Hello, <your-input>，偶数个字符就输出<your-input>, World!。

于是，构建三个节点：意图识别节点、Hello节点和World节点。

每个节点都采用一个统一的传输格式，也就是包括：

  • 意图标志位的branch
  • 输入的text
  • 输出的answer

于是，我们先定义数据结构：

1
2
3
4
class State(TypeDict):
    text: str
    branch: int
    answer: str

节点就像这样：

1
2
3
4
5
6
7
8
9
def detect_node(state: State) -> State:
    """检测节点"""
    return {"branch": len(state["text"]) % 2, "answer": ""}
def hello_node(state: State) -> State:
    """Hello节点"""
    return {"answer": f"Hello, {state['text']}!"}"}
def world_node(state: State) -> State:
    """World节点"""
    return {"answer": f"{state['text']}, World!"}

然后就是建立LangGraph：

1
2
3
4
5
6
7
8
9
10
11
12
13
def build() -> StateGraph:
    sg = StateGraph(State)
    sg.add_node("intent", detect_intent)
    sg.add_node("hello", hello_node)
    sg.add_node("world", world_node)
    sg.add_conditional_edges(
        "intent", lambda s: "hello" if s["branch"] else "world"
    )
    sg.add_edge(START, "intent")
    sg.add_edge("hello", END)
    sg.add_edge("world", END)
    graph = sg.compile()
    return graph

MCP Server

既然有了意图识别的基础，服务器端就可以建立起来了：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import asyncio
from mcp.server.fastmcp import FastMCP

# 起一个最简单的 MCP 服务器
mcp = FastMCP(name = "HappyServer")

# 定义一个工具
@mcp.tool()
def hello(text: str) -> str:
    """Import workflow and use it"""
    graph = build_graph()
    response: dict = graph.invoke({"text": text})
    return response.get("answer", "result")

if __name__ == "__main__":
    # 通过 stdio 运行（直接 python server.py 即可被客户端拉起）
    asyncio.run(mcp.run_streamable_http_async())

P.S.：当然，你完全可以把这个build_graph方法放在另一个文件中，然后import进来，效果当然是一样的。

MCP Client

这个部分可以参考上一篇文章，代码甚至都没有变化。

运行

首先，我们将server启动起来，只需要运行MCP Server那个文件即可；

然后，我们运行client，同样运行MCP Client文件即可。

于是，我们就可以看到这些：

  • 当我们输入LangGraph的时候，由于是奇数个字符，所以走到了Hello, LangGraph!的逻辑里面；
  • 当我们输入vLLM的时候，由于是偶数个字符，所以走到了vLLM, World!的逻辑里面；

总结以及需要注意的小东西

这样，我们就将LangGraph引入了MCP，也就可以实现更复杂的逻辑了。

但是需要注意的是，MCP不能返回字符串以外的内容。比如，在MCP Server中，我们不能直接输出LangGraph默认的TypeDict返回值，这样就会报错：

1
2
3
4
Error executing tool hello: 1 validation error for helloOutput
result
  Input should be a valid string [type=string_type, input_value={'text': 'LangGraph', 'br...r': 'Hello，LangGraph'}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.11/v/string_type

所以在返回的时候，我使用的是：

1
response.get("answer", "result")

在获取结果的时候，如果没有拿到answer中的字符串，那就直接返回一个result算了。

而如果直接返回response，那么返回结果就成了：

1
2
3
4
5
{
    "branch": 0,
    "text": "LangGraph",
    "answer": "Hello, LangGraph!"
}

这样，MCP Client在执行过程中，会使用send_request方法发送CallToolRequest，并在发送请求后立即使用model_validate方法检查返回值是否符合要求。显然，TypeDict与str之间还是有着显著的差异，所以不通过检查。

因此，在使用MCP的时候，所有的内容都只能使用str传参。