LangGraph 接入MCP

文章目录

• • 1.MCP
  • 2.工作机制
  • 3.MCP的两种实现方式
  • 4.Agent接入MCP服务
  • 5.手写MCP服务

1.MCP

MCP即Model Context Protocol（模型上下文协议），于 2024 年 11 月由 Claude 大模型的公司 Anthropic 推出并开源，提供统一的通信接口，旨在解决大模型与外部数据源、工具及系统集成时的碎片化问题，MCP 可以理解为AI 世界的 USB-C 接口，它的核心目标是让 AI 模型（如 ChatGPT、Claude 等）能够像通用遥控器一样，通过统一协议连接各类数据源和工具，实现高效、安全的跨系统协作。

MCP可以解决Function calling的弊端，Function calling在使用时需要构建一个外部函数作为中介，一边接收大模型发出的请求，另一边负责调用外部工具，通过这种方式使大模型能够间接实现对外部工具的调用。 比如当查询天气时，大模型调用外部工具的function calling过程如下： 此时需要编写外部函数，外部函数往往需要上百行代码，而且还需要设计提示词模版，才能提高Function calling响应的准确率。

而MCP能在Agent开发过程中，让大模型更加便捷的调用外部工具。MCP首先统一了Function calling的运行规范。MCP把大模型运行环境称作 MCP Client，也就是MCP客户端，同时，把外部函数运行环境称作MCP Server，也就是MCP服务器，然后统一了MCP客户端和服务器的运行规范，并且要求MCP客户端和服务器之间，也统一按照某个既定的提示词模板进行通信。

2.工作机制

以下案例给⼤模型提供路线规划的能⼒。 以⾼德地图提供的MCP服务为例，⾼德地图开放平台提供了MCP接⼊介绍： https://lbs.amap.com/api/mcp-server/summary。 只需要在⾼德开放平台注册账号，登录后，在左侧菜单中选择“应⽤管理”->“我的应⽤”，然后创建新应⽤，就可以申请api_key

from langchain_openai import ChatOpenAI
import datetime
from langchain.tools import tool

llm = ChatOpenAI(
model="qwen-plus",
base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
openai_api_key="API_KEY",
streaming=True
)

# 定义工具
@tool(description="规划行车路线")
def get_route_plan(origin_city: str, target_city: str):
"""规划行车路线
Args:
origin_city: 出发城市
target_city: 目标城市
"""
result = f"从城市 {origin_city} 出发，到目标城市 {target_city} ,只需意念，三分钟即可到达。"
print(">>>> get_route_plan >>>>>" + result)
return result

# 大模型绑定工具
llm_with_tools = llm.bind_tools([get_route_plan])

# 工具容器
all_tools = {"get_route_plan": get_route_plan}

# 准备查询和消息列表
query = "帮我规划一条从广州到新疆的自驾路线"
messages = [query]

# 获取工具调用请求
ai_msg = llm_with_tools.invoke(messages)
print(ai_msg)
messages.append(ai_msg)

# 打印并处理工具调用
print(ai_msg.tool_calls)
if ai_msg.tool_calls:
for tool_call in ai_msg.tool_calls:
selected_tool = all_tools[tool_call["name"].lower()]
tool_msg = selected_tool.invoke(tool_call)
messages.append(tool_msg)

# 获取最终响应
print("\\n流式输出结果：")
for chunk in llm_with_tools.stream(messages):
if chunk.content:
print(chunk.content, end="", flush=True)

从结果可以看出高德提供的 MCP 服务数据可信度更高，所以被大模型直接采用它的数据，自己提供的意念传送相关数据缺乏可靠性，大模型参考了，但是最终没有采纳。可以知道工具在处理具体问题时存在的缺陷，是否调用工具完全由大模型自主决定，无论工具返回什么，最终大模型输出的内容还是自己说了算，至于工具在执行过程中的具体操作，大模型则不干预。

3.MCP的两种实现方式

分别是基于标准输入输出（stdio）的本地进程通信和基于 HTTP/SSE 的网络通信，以高德地图的 MCP 服务为例，其开放平台的说明中给出了两种接入该服务的配置方式。 sse方式

{
"mcpServers": {
"amap-amap-sse": {
"url": "https://mcp.amap.com/sse?key=key"
}
}
}

stdio方式

"amap-maps": {
"command": "npx",
"args": [
"-y",
"@amap/amap-maps-mcp-server"
],
"env": {
"AMAP_MAPS_API_KEY": "key"
}
}

两种方式区别 实际使用时，本地开发常用stdio方式，生产环境的分布式场景则更多使用HTTP/SSE方式。

4.Agent接入MCP服务

首先安装依赖

pip install langchain-mcp-adapters

from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent
from langchain_community.chat_models import ChatTongyi
import asyncio

llm = ChatTongyi(
model="qwen-plus",
api_key="API_KEY",
)

async def main():
# 配置高德地图MCP客户端，使用SSE方式连接
client = MultiServerMCPClient(
{
"amap-maps-sse": {
"url": "https://mcp.amap.com/sse?key=xxxxxxxx",
"transport": "sse"
},
# "amap-maps": {
# "command": "npx",
# "args": [
# "-y",
# "@amap/amap-maps-mcp-server"
# ],
# "env": {
# "AMAP_MAPS_API_KEY": "xxxxxxxx"
# },
# "transport":"stdio"
# }
}
)
try:
tools = await client.get_tools()
agent = create_react_agent(
model=llm,
tools=tools
)
response = await agent.ainvoke(
{"messages": [{"role": "user", "content": "帮我规划一条从广州到新疆的自驾路线"}]}
)
return response
except Exception as e:
print(f"发生错误: {e}")
return None

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

5.手写MCP服务

MCP服务

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("roymcpdemo")

@mcp.tool()
def add(a: int, b: int) –> int:
"""Add two numbers together."""
print(f"roy mcp demo called : add({a}, {b})")
return a + b

@mcp.tool()
def weather(city: str):
"""获取某个城市的天气
Args:
city: 具体城市
"""
return "城市" + city + ",今天天气不错"

@mcp.resource("greeting://{name}")
def greeting(name: str) –> str:
"""Greet a person by name."""
print(f"roy mcp demo called : greeting({name})")
return f"Hello, {name}!"

if __name__ == "__main__":
# 以sse协议暴露服务。
mcp.run(transport='sse')

# 以stdio协议暴露服务。
# mcp.run(transport='stdio')

运行服务端

MCP客户端

from mcp import ClientSession
import mcp.types as types

SERVER_URL = "http://127.0.0.1:8000/sse"

async def handle_sampling_message(message: types.CreateMessageRequestParams) –> types.CreateMessageResult:
print(f"sampling message: {message}")
return types.CreateMessageResult(
role="assistant",
content=types.TextContent(
type="text",
text="Hello,world! from model"
),
model="qwen-plus",
stopReason="endTurn"
)

async def run():
from mcp.client.sse import sse_client
async with sse_client(SERVER_URL) as (read, write):
async with ClientSession(read, write, sampling_callback=handle_sampling_message) as session:
await session.initialize()

prompts = await session.list_prompts()
print(f"prompts: {prompts}")

tools = await session.list_tools()
print(f"tools: {tools}")

resources = await session.list_resources()
print(f"resources: {resources}")

result = await session.call_tool("weather", {"city": "北京"})
print(f"result: {result}")

if __name__ == "__main__":
import asyncio

asyncio.run(run())

调用MCP服务端