使用 LangGraph 构建本地 MCP 工具：一步一步指南

想象一下，你是否曾经梦想过拥有一个完全属于自己的 AI 助手，能够满足你个性化的需求，例如查询天气、获取新闻、甚至是控制智能家居设备？现在，借助 MCP (Model Context Protocol) 协议 和强大的 LangGraph 框架，这个梦想触手可及。本文将带你一步一步地了解如何创建和使用本地 MCP 工具，让你能够完全掌控你的 AI 超能力。

理解 MCP (Model Context Protocol) 的本质

MCP (Model Context Protocol)，即模型上下文协议，是一种允许语言模型通过结构化通信访问各种工具（例如 APIs、函数或脚本）的协议。你可以将它想象成一种桥梁，连接了 AI 模型和外部世界，使得模型可以像调用自身内部函数一样调用这些工具。这对于构建智能、多功能的 AI 代理至关重要。

更具体地说，MCP 定义了一种标准化的格式，用于在语言模型和工具之间传递信息。这种格式包括：

请求 (Request): 模型向工具发送的指令，包含工具需要执行的任务以及必要的参数。
响应 (Response): 工具执行任务后返回的结果，包含了模型所需的信息。

通过这种标准化的通信方式，不同的语言模型可以轻松地与各种类型的工具进行交互，而无需针对每种工具进行特殊的集成。这大大提高了 AI 系统的灵活性和可扩展性。

例如，假设你想让 AI 模型查询北京的天气。使用 MCP 协议，你可以构建一个天气查询工具，并将其暴露给模型。模型通过 MCP 向工具发送一个包含“查询北京天气”的请求，工具则会返回包含北京天气信息的响应。

本地 MCP 服务器的优势

为什么我们需要构建一个本地 MCP 服务器 呢？原因有很多，主要包括以下几点：

高度可插拔性： 本地服务器允许你轻松地添加或删除工具（函数），而无需修改核心 AI 模型。你可以根据自己的需求，随时定制和更新你的工具集。例如，你可以添加一个用于总结新闻文章的工具，或者添加一个用于控制智能灯泡的工具。
无与伦比的实验灵活性： 本地环境是探索新想法的完美场所。你可以自由地测试各种有趣的概念，例如个性化机器人、搜索助手或游戏代理，而无需担心云服务的限制或成本。例如，你可以创建一个本地 AI 代理，用于自动回复你的邮件，或者创建一个 AI 助手，用于帮助你进行市场调研。
灵活的技术栈： 本地 MCP 服务器可以与 LangChain、LangGraph 或你自己的定制代理无缝集成。你可以选择最适合你需求的框架和工具，构建你理想的 AI 系统。例如，你可以使用 LangChain 来构建一个对话式 AI 代理，并使用 LangGraph 来 orchestrate 不同的工具。

构建本地 MCP 服务器，意味着你获得了对 AI 工具的完全控制权，可以自由地探索和实验，打造个性化的 AI 解决方案。

LangGraph：构建复杂 AI 工作流的利器

在构建本地 MCP 工具 时，LangGraph 框架可以发挥关键作用。LangGraph 是一个用于构建复杂 AI 工作流的强大工具，它允许你将不同的工具和模型连接起来，构建一个完整的 AI 解决方案。

具体来说，LangGraph 提供了以下关键特性：

图结构： LangGraph 使用图结构来表示 AI 工作流。图中的节点可以是工具、模型或其他组件，而边则表示数据流。这种图结构使得 AI 工作流的可视化和管理变得更加容易。
状态管理： LangGraph 可以管理 AI 工作流的状态。状态可以包括模型的状态、工具的状态以及中间结果。通过状态管理，LangGraph 可以确保 AI 工作流的正确执行，并支持复杂的控制流。
并发执行： LangGraph 支持并发执行。这意味着多个工具和模型可以同时运行，从而提高 AI 工作流的效率。

通过使用 LangGraph，你可以将不同的 MCP 工具 连接起来，构建一个复杂的 AI 工作流，例如：

用户发起一个请求，例如“查找关于人工智能的最新新闻”。
LangGraph 首先将用户的请求发送给一个新闻搜索工具，该工具使用 MCP 协议从互联网上搜索相关的新闻文章。
然后，LangGraph 将搜索到的新闻文章发送给一个摘要工具，该工具使用 MCP 协议对新闻文章进行摘要。
最后，LangGraph 将摘要后的新闻文章返回给用户。

这个例子展示了如何使用 LangGraph 将不同的 MCP 工具 连接起来，构建一个完整的 AI 解决方案。

Python 实现本地 MCP 工具：手把手教程

现在，让我们开始动手构建一个简单的本地 MCP 工具，并将其与 LangGraph 集成。我们将使用 Python 作为编程语言。

1. 安装必要的库:

首先，确保你已经安装了 Python 和 pip。然后，使用 pip 安装以下库：

pip install fastapi uvicorn pydantic langchain langgraph

fastapi 和 uvicorn 用于构建和运行 Web API。
pydantic 用于数据验证和序列化。
langchain 和 langgraph 用于构建 AI 代理和工作流。

2. 创建一个简单的 MCP 工具：天气查询工具

创建一个名为 weather_tool.py 的文件，并添加以下代码：

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import uvicorn
import random

app = FastAPI()

class WeatherRequest(BaseModel):
    city: str

class WeatherResponse(BaseModel):
    city: str
    temperature: int
    condition: str

@app.post("/weather")
async def get_weather(request: WeatherRequest):
    """
    模拟天气查询 API。
    """
    city = request.city
    # 模拟天气数据
    temperature = random.randint(10, 30)
    conditions = ["Sunny", "Cloudy", "Rainy", "Snowy"]
    condition = random.choice(conditions)

    # 返回天气信息
    return WeatherResponse(city=city, temperature=temperature, condition=condition)

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

这个简单的工具模拟了一个天气查询 API。它接收一个包含城市名称的 WeatherRequest，并返回一个包含城市、温度和天气状况的 WeatherResponse。注意，这里我们只是模拟了天气数据，实际应用中需要调用真实的天气 API。

3. 运行 MCP 工具服务器:

在终端中运行以下命令：

python weather_tool.py

这将启动一个运行在 http://0.0.0.0:8000 的 Web 服务器。

4. 使用 LangGraph 调用 MCP 工具:

创建一个名为 langgraph_example.py 的文件，并添加以下代码：

from langchain.chains import APIChain
from langchain.prompts import PromptTemplate
from langchain.llms import OpenAI # 需要设置 OpenAI API 密钥
from langgraph.graph import StateGraph, END
from typing import TypedDict, Dict, Any

import os

os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY" # 替换为你的 OpenAI API 密钥

# 定义状态类型
class AgentState(TypedDict):
    messages: list[dict[str, Any]]
    steps: list[str]

# 创建 LangChain 工具
api_docs = """
This is an API that returns the weather for a given city.
Base URL: http://0.0.0.0:8000
Endpoint: /weather
Request:
  - city (string): The city to get the weather for.
Response:
  - city (string): The city requested.
  - temperature (integer): The temperature in Celsius.
  - condition (string): The weather condition.
"""

# 创建 LangChain APIChain
prompt = PromptTemplate.from_template("""
You are a helpful assistant that answers questions about the weather.
Use the following API to answer the question:
{api_docs}

Begin!

Question: {question}
Answer: """)

llm = OpenAI(temperature=0) # 初始化 OpenAI LLM

api_chain = APIChain.from_llm_and_api_docs(
    llm,
    api_docs,
    prompt=prompt,
    verbose=True # 打印详细日志
)

# 定义节点
def call_api(state):
    question = state['messages'][-1]['content'] # 获取用户提问
    result = api_chain.run(question)
    return {"messages": [{"role": "assistant", "content": result}]}

def generate_response(state):
    """
    生成最终回复。
    """
    final_response = "我不太确定，但是根据天气预报..." + state['messages'][-1]['content']
    return {"messages": [{"role": "assistant", "content": final_response}]}

# 定义图
graph = StateGraph(AgentState)
graph.add_node("api_call", call_api)
graph.add_node("generate_response", generate_response)

# 定义边
graph.add_edge("api_call", "generate_response")
graph.set_entry_point("api_call")
graph.add_edge("generate_response", END)

# 编译图
chain = graph.compile()

# 运行链
inputs = {"messages": [{"role": "user", "content": "北京的天气怎么样？"}]}
result = chain.invoke(inputs)

print(result)

这个例子展示了如何使用 LangGraph 调用我们创建的本地天气查询 MCP 工具。

首先，我们定义了 AgentState，它包含了对话历史记录 messages 和执行步骤 steps。
然后，我们创建了一个 APIChain，它使用 OpenAI LLM 和 API 文档来调用 MCP 工具。
接下来，我们定义了两个节点：api_call 和 generate_response。api_call 节点负责调用 MCP 工具，generate_response 节点负责生成最终回复。
最后，我们使用 StateGraph 创建一个图，并将节点和边连接起来。

5. 运行 LangGraph 示例:

在终端中运行以下命令：

python langgraph_example.py

确保你的 OpenAI API 密钥已正确设置。

运行后，你将看到 LangGraph 调用本地天气查询 MCP 工具 并生成回复。这个例子演示了如何将 LangGraph 与本地 MCP 工具 集成，构建一个简单的 AI 工作流。

总结与展望

通过本文的介绍，你应该已经了解了 MCP (Model Context Protocol) 的基本概念，以及如何使用 LangGraph 框架构建和调用本地 MCP 工具。利用这种方法，你可以轻松地扩展 AI 模型的 capabilities，并构建出更智能、更强大的 AI 应用。

未来，MCP 工具 将会在 AI 领域扮演越来越重要的角色。我们可以期待更多类型的工具被暴露给 AI 模型，例如：

数据分析工具： 用于分析用户数据，并生成个性化的推荐。
自动化工具： 用于自动化重复性的任务，例如发送邮件或更新社交媒体。
物理世界交互工具： 用于控制机器人、智能家居设备或其他物理设备。

随着 MCP 协议 的不断发展和完善，我们可以期待一个更加智能、更加便捷的 AI 世界。

使用 LangGraph 构建本地 MCP 工具：一步一步指南