从理论到实践：使用Model Context Protocol构建多工具AI代理的完整指南

类比

我们都熟悉《Kaun Banega Crorepati（KBC）》节目中的“Phone a Friend（打电话求助）”环节。这是印度版的《Who Wants to Be a Millionaire?》。

现在，想象一下如果 KBC 节目诞生于“电话尚未发明”的时代。

在没有电话的世界里：
如果节目想让选手“打电话”求助朋友，就必须为每一位求助的朋友进行复杂的物理准备：
* 将朋友请到演播室现场。
* 准备独立的房间，并安装麦克风、耳机和布线。
* 对每一位选手的每一位朋友都重复这套流程。

有了电话之后：
一切都改变了。KBC 不再需要复杂的物理搭建。选手只需：
* 拿起电话。
* 拨打朋友的号码。
* 直接通话。

无需专用房间，也无需专门布置。这正是 Model Context Protocol（MCP）为 LLM/AI 应用带来的变革。

没有 MCP 时：
* 每个外部工具都需要进行自定义、一次性的集成。
* 没有标准化的方法让模型与工具“对话”。
* 所有组件都需要开发者手动“接线”到一起。

有了 MCP 之后：
* 任何兼容 MCP 的 AI 模型都能“拨号”调用任何兼容 MCP 的工具。
* 模型只需知道工具的“模式”（schema），就像拥有一个电话号码。
* 不再需要为每个工具重复开发专用集成。
* 所有交互都通过同一个标准协议完成。

引言

在 AI 与大语言模型快速演进的今天，核心挑战之一是如何将这些强大的模型与真实世界的数据和工具无缝连接。由 Anthropic 于 2024 年提出的 Model Context Protocol（MCP），作为一个开放标准，正在从根本上改变 AI 应用与外部系统交互的方式。

在本指南中，我们将通过一个实践项目，构建一个能够获取实时天气数据并执行网页搜索的多工具 AI Agent。无论你是开发者、AI 爱好者，还是对 AI 集成架构感兴趣的研究者，本文都将帮助你掌握构建基于 MCP 的应用的实用方法。

为什么选择 MCP？

MCP 充当了 AI 模型与其所需工具之间的标准化桥梁，使我们能够构建能与外界进行复杂、安全交互的智能体。

通过消除集成孤岛，并标准化模型与外部系统之间的上下文共享方式，MCP 使得 AI 更加互联、功能更强大，也更易于企业级部署。MCP 主要解决了以下问题：

通用语言
- MCP 之前：每个工具都有其独特的与 AI 模型交互的接口和方式。
- MCP 之后：所有工具使用同一种共享的协议语言进行通信，沟通变得简单一致。
轻松集成
- MCP 之前：开发者需要为每个应用单独开发和维护连接不同工具的适配器。
- MCP 之后：一个标准化的 MCP 客户端可以连接所有兼容 MCP 的工具，实现“一次编写，到处连接”。
安全的工具访问
- MCP 之前：AI 模型直接、无差别地访问外部工具和数据可能带来安全风险。
- MCP 之后：MCP 协议层可以集成认证和授权机制，让 AI 能够安全、受控地使用多种工具。
可复用的工具
- MCP 之前：为特定模型（如 ChatGPT 插件）开发的工具通常无法在其他模型或平台上使用。
- MCP 之后：任何遵循 MCP 标准的工具都能被任何兼容 MCP 的模型或平台调用，极大地提高了工具的可复用性。
更快的创新
- MCP 之前：社区在集成工作上重复“造轮子”，拖慢了整体创新进度。
- MCP 之后：共享的标准降低了集成门槛，开发者可以更专注于构建新工具和新应用，从而加速整个生态的创新。

MCP（Model Context Protocol）的关键组件

从理论到实践：使用Model Context Protocol构建多工具AI代理的完整指南

Architecture of MCP (Model Context Protocol)

1. MCP Host

这是运行 AI 应用或智能体的主环境，也是用户进行交互的界面（例如 Claude Desktop、集成开发环境 IDE、自定义的对话应用）。
Host 负责编排一个或多个 MCP Client，管理工作流程，并决定何时以及如何调用各个 MCP Server 提供的工具。

2. MCP Client

MCP Client 运行在 Host 应用中，负责与一个或多个 MCP Server 进行通信。其主要职责包括：
* 与 MCP Server 建立连接。
* 向 Server 请求其暴露的可用能力（工具、资源等）列表。
* 将 AI 模型的意图或用户请求，翻译成结构化的 MCP 协议消息，并发送给对应的 Server。
* 接收、解析 Server 返回的响应，并将其传递给 Host 或 AI 模型。

3. MCP Server

MCP Server 提供具体的功能，对外开放对不同数据源和工具的访问。每个工具（如数据库、API、文件系统）都可以实现为一个 MCP Server，一个 Server 也可以包含多个相关工具。其核心职责包括：
* 监听来自 Client 的请求。
* 执行请求所对应的函数（如调用 API、查询数据库）或检索数据。
* 将执行结果构造成结构化的数据（通常是 JSON 格式）返回给 Client。

每个 Server 会向 Client 公开三类核心能力：
* Tools（工具）：允许 LLM 执行动作的函数（例如：执行搜索、发送邮件、更新数据库）。
* Resources（资源）：LLM 可以读取或访问的静态或动态数据（例如：文档、知识库条目、用户配置信息）。
* Prompts（提示词）：可复用的提示模板或工作流，用于引导模型生成特定类型或格式的输出。

4. Protocol（协议）

MCP 使用 JSON-RPC 2.0 作为 Client 与 Server 之间通信的数据交换格式。
它定义了标准的消息类型、语义以及针对 “tools”、“resources”、“prompts” 等核心概念的交互原语。
生命周期管理：MCP 协议处理连接的建立、初始的能力协商（确认 Server 支持哪些功能）以及连接的正常关闭。
安全性：协议支持集成 OAuth 等认证机制，为 Client 与 Server 之间的交互提供授权保障。

5. Transport Layer（传输层）

抽象了消息的实际传输方式。MCP 支持多种传输机制以适应不同场景。
两种常见的传输方式是：
* STDIO（标准输入输出）：用于本地、同步运行的 Server，通信通过进程的标准输入输出流进行。
* Server-Sent Events (SSE)：用于远程或需要支持流式、实时更新的 Server。

我们要构建什么

在这个实践项目中，我们将构建一个基于 LangGraph 的智能体（Agent），它具备以下能力：
* 集成 MCP 工具（web-search 和 weather）来执行外部动作。
* 使用 OpenAI 的 LLM 进行意图判断、决策路由和最终答案生成。
* 通过 LangGraph 的有状态图（Stateful Graph）来处理多轮对话和复杂工作流。
* 架构易于扩展，可以方便地加入更多 MCP 工具或构建更复杂的推理图。

通过这个架构，你将能够打造出能够自主推理、调用工具并动态响应用户需求的 AI 助手。

Model Context Protocol Implementation

Flow between User Query → Final Response using MCP

让我们开始动手。

项目结构

完整的项目代码可在 GitHub 仓库找到：https://github.com/alphaiterations/mcp-server

mcp-server/
├── mcp_server.py          # MCP server 实现 (基于 FastMCP)
├── mcp_client.py          # MCP client 与 LangGraph agent 集成
├── mcp-server.ipynb       # 用于测试的 Jupyter notebook
├── requirements.txt       # Python 项目依赖
├── .env                   # 环境变量文件 (存放 API 密钥)
└── README.md

1. 安装与环境配置

1.1 克隆仓库

在终端中运行以下命令：

git clone https://github.com/alphaiterations/mcp-server
cd mcp-server

1.2 安装依赖

在项目根目录下，运行：

pip install -r requirements.txt

1.3 配置环境变量

在项目根目录下创建一个名为 .env 的文件，并填入以下内容：

# .env
# OpenAI API Key (必需)
OPENAI_API_KEY=your_openai_api_key_here

# Serper API Key (用于网页搜索，必需)
SERPER_API_KEY=your_serper_api_key_here

如何获取 API Key：
* OpenAI: 访问 https://platform.openai.com/api-keys 创建。
* Serper: 访问 https://serper.dev/ 注册，它提供免费的额度可供测试。

2. 搭建 MCP Server

我们将在 mcp_server.py 中搭建 MCP Server。这个脚本创建了一个包含两个实用工具的 MCP Server：

get_weather: 获取指定城市的实时天气信息。
web_search: 执行针对 AI 优化过的 Google 搜索。

这两个工具都内置了输入验证、错误处理和结构化 JSON 输出，非常适合与 LLM 集成。

2.1 导入必要的库

我们首先导入所需的模块：

# mcp_server.py
from fastmcp import FastMCP
from pydantic import BaseModel
import os
import requests
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件中的环境变量

2.2 API 配置

从环境变量中读取所需的 API 密钥。OPENWEATHER_API_KEY 被设为可选，因为我们的天气工具实现了一个免费的备用数据源。

# mcp_server.py
# ---------------- API 配置 ----------------
SERPER_API_KEY = os.getenv("SERPER_API_KEY")
OPENWEATHER_API_KEY = os.getenv("OPENWEATHER_API_KEY", "")  # 可选，我们使用免费的天气 API 作为备用

2.3 创建 MCP Server 实例

使用 FastMCP 初始化一个 Server：

# mcp_server.py
mcp = FastMCP("multi-tool-server")

这将初始化一个名为 “multi-tool-server” 的 MCP Server 实例。接下来，我们使用 @mcp.tool() 装饰器注册的函数，都会自动成为该 Server 对外提供的、可被 AI Agent 调用的工具。

2.4 天气工具

# mcp_server.py
@mcp.tool()
def get_weather(input: WeatherInput):
    """Get current weather for a city using WeatherAPI.com (free tier)"""
    try:
        # Using WeatherAPI.com free tier (no API key needed for basic usage)
        # Alternative: OpenWeatherMap if you have API key
        url = f"https://wttr.in/{input.city}?format=j1"
        response = requests.get(url, timeout=5)

        if response.status_code == 200:
            data = response.json()
            current = data["current_condition"][0]
            location = data["nearest_area"][0]

            weather_info = {
                "location": f"{location['areaName'][0]['value']}, {location['country'][0]['value']}",
                "temperature": f"{current['temp_C']}°C / {current['temp_F']}°F",
                "condition": current["weatherDesc"][0]["value"],
                "humidity": f"{current['humidity']}%",
                "wind": f"{current['windspeedKmph']} km/h",
                "feels_like": f"{current['FeelsLikeC']}°C"
            }
            return weather_info
        else:
            return {"error": f"Could not fetch weather for {input.city}"}
    except Exception as e:
        return {"error": f"Weather API error: {str(e)}"}

此代码注册了一个通过 MCP 协议可访问的天气查询工具。其核心实现如下：

API 选择：使用无需 API 密钥的 wttr.in 免费天气服务。
数据处理：从 API 响应中提取温度、湿度、风速等关键信息，并格式化为对人类友好的字段。
错误处理：使用 try/except 块包裹网络请求，确保在 API 调用失败时能安全地返回结构化的错误信息。

输入/输出示例：

Input:  {"city": "Mumbai"}
Output: {
    "location": "Mumbai, India",
    "temperature": "28°C / 82°F",
    "condition": "Partly cloudy",
    "humidity": "70%",
    "wind": "15 km/h"
}

2.5 Web 搜索工具

# mcp_server.py

@mcp.tool()
def web_search(input: WebSearchInput):
    """Search the web using Serper API"""
    if not SERPER_API_KEY:
        return {"error": "SERPER_API_KEY not configured"}

    try:
        url = "https://google.serper.dev/search"
        headers = {
            "X-API-KEY": SERPER_API_KEY,
            "Content-Type": "application/json"
        }
        payload = {"q": input.query, "num": 5}

        response = requests.post(url, json=payload, headers=headers, timeout=10)

        if response.status_code == 200:
            data = response.json()
            results = []

            # Extract organic results
            for item in data.get("organic", [])[:5]:
                results.append({
                    "title": item.get("title"),
                    "link": item.get("link"),
                    "snippet": item.get("snippet")
                })

            # Include knowledge graph if available
            knowledge = data.get("knowledgeGraph", {})

            return {
                "query": input.query,
                "results": results,
                "knowledge_graph": knowledge.get("description", "") if knowledge else None
            }
        else:
            return {"error": f"Serper API error: {response.status_code}"}
    except Exception as e:
        return {"error": f"Web search error: {str(e)}"}

此函数注册了一个由 Serper API 驱动的网络搜索工具。实现要点包括：

前置校验：在发起请求前检查 SERPER_API_KEY 是否已配置。
请求与限制：通过 POST 请求发送查询，并限制返回最多 5 条结果。
结果解析：从 API 响应中解析出有机搜索结果的标题、链接和摘要片段。如果存在知识图谱信息，也一并提取。
错误处理：所有潜在异常均被捕获，并以结构化的 JSON 格式返回错误信息。

输入/输出示例：

Input:  {"query": "latest news about AI"}
Output: {
    "query": "latest news about AI",
    "results": [
        {
            "title": "AI Breakthrough...",
            "link": "https://...",
            "snippet": "..."
        }
    ]
}

2.6 Server 入口

# mcp_server.py
if __name__ == "__main__":
    print("🔥  MCP server running...")
    mcp.run()

当脚本被直接执行时，这段代码会启动 MCP 服务器。服务器启动后，get_weather 和 web_search 等已注册的工具即可被以下客户端调用：

AI Agents
LangChain / LLM 框架
任何兼容 MCP 协议的客户端

3. 搭建 MCP Client

3.1 依赖与环境

# mcp_client.py

import subprocess, json, sys, threading, time
from langgraph.graph import StateGraph, END
from typing import TypedDict, Any, Dict
from openai import OpenAI
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()

客户端代码导入以下关键模块：

LangGraph：用于构建基于状态（State）的智能体工作流。
OpenAI：负责与大型语言模型（LLM）进行交互和推理。
subprocess / threading：用于启动 MCP 服务器子进程并与之进行通信。
dotenv / os：安全地从 .env 文件加载环境变量（如 API 密钥）。

3.2 初始化 OpenAI Client

# mcp_client.py
client = OpenAI(api_key=os.getenv("OPEN_AI_KEY"))

使用从环境变量 OPEN_AI_KEY 中获取的 API 密钥，初始化 OpenAI 客户端。

3.3 启动 MCP Server

# mcp_client.py
server = subprocess.Popen(
    [sys.executable, "mcp_server.py"],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1
)

print("🔗  Connected to MCP server")

此段代码将 mcp_server.py 作为一个子进程启动，使其在后台运行并暴露之前定义的 get_weather 与 web_search 等工具。通过管道（stdin/stdout/stderr）建立与子进程的通信通道。

为了非阻塞地等待服务器启动完成，我们使用一个单独的线程来读取服务器的启动日志：

# mcp_client.py
def read_startup():
    while True:
        line = server.stderr.readline()
        if not line:
            break
        if "Starting MCP server" in line:
            break

startup_thread = threading.Thread(target=read_startup, daemon=True)
startup_thread.start()
time.sleep(1)

3.4 MCP 通信辅助方法

3.3 实现 MCP 客户端通信

我们首先定义两个核心函数，用于通过 MCP 的 JSON-RPC 协议与服务器进行通信：一个用于发送请求并获取响应，另一个用于发送无需响应的通知。

# mcp_client.py
def send_request(method: str, params: Dict = None):
    global request_id
    request_id += 1
    req = {"jsonrpc": "2.0", "id": str(request_id), "method": method}
    if params:
        req["params"] = params
    server.stdin.write(json.dumps(req) + "n")
    server.stdin.flush()
    return server.stdout.readline().strip()

def send_notification(method: str, params: Dict = None):
    req = {"jsonrpc": "2.0", "method": method}
    if params:
        req["params"] = params
    server.stdin.write(json.dumps(req) + "n")
    server.stdin.flush()

建立连接后，客户端需要向服务器发送初始化请求，并通知服务器初始化已完成。

# mcp_client.py
init_response = send_request("initialize", {
    "protocolVersion": "2024-11-05",
    "capabilities": {},
    "clientInfo": {"name": "mcp-client", "version": "1.0.0"}
})
send_notification("initialized")

3.4 调用 MCP 工具

为了调用服务器上注册的工具，我们实现一个专门的函数。该函数构造一个 tools/call 请求，发送给服务器，并解析返回的结果。

# mcp_client.py
def call_mcp_tool(tool: str, args: Dict):
    global request_id
    request_id += 1
    req = {
        "jsonrpc": "2.0",
        "id": str(request_id),
        "method": "tools/call",
        "params": {"name": tool, "arguments": {"input": args}},
    }
    server.stdin.write(json.dumps(req) + "n")
    server.stdin.flush()
    response = server.stdout.readline().strip()
    try:
        resp_data = json.loads(response)
        return resp_data.get("result") or f"Error: {resp_data.get('error')}"
    except:
        return response

此函数是代理（Agent）能力扩展的关键，使其能够通过 MCP 协议使用外部工具。

3.5 实现代理逻辑

代理的核心是状态管理和决策。我们首先使用 TypedDict 定义代理的状态结构。

# mcp_client.py
class S(TypedDict):
    msg: str
    tool_result: Any
    result: str

代理的逻辑主要由两个函数实现：工具路由和回答生成。

1. 工具路由 (route_request)

这个函数负责分析用户请求，并决定是否需要调用工具、调用哪个工具以及使用什么参数。

# mcp_client.py
# ---------------- AGENT LOGIC ----------------
def route_request(state: S):
    """Use LLM to determine which tool to use and extract parameters"""

    # 构建提示词，要求 LLM 分析请求并做出路由决策
    routing_prompt = f"""Analyze this user request and determine which tool to use:

User request: {state["msg"]}

Available tools:
1. get_weather - Get current weather for a city. Requires: city name
2. web_search - Search the web for information. Requires: search query

Respond in JSON format with:
{{
    "tool": "get_weather" | "web_search" | "none",
    "parameters": {{"city": "..."}} or {{"query": "..."}} or null,
    "reasoning": "brief explanation"
}}

Only use tools if clearly needed. For general conversation, use "none"."""

    # 调用 LLM 获取决策
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a tool routing assistant. Analyze user requests and determine which tool to use. Always respond with valid JSON."},
            {"role": "user", "content": routing_prompt}
        ],
        temperature=0.3,
        response_format={"type": "json_object"}
    )

    try:
        routing_decision = json.loads(response.choices[0].message.content)
        tool = routing_decision.get("tool")
        params = routing_decision.get("parameters")

        print(f"🤖 Routing decision: {routing_decision.get('reasoning')}")

        # 根据决策调用相应的 MCP 工具
        if tool == "get_weather" and params:
            result = call_mcp_tool("get_weather", params)
            return {"msg": state["msg"], "tool_result": result}

        elif tool == "web_search" and params:
            result = call_mcp_tool("web_search", params)
            return {"msg": state["msg"], "tool_result": result}

        else:
            # 无需调用工具，进入直接对话流程
            return {"msg": state["msg"], "tool_result": None}

    except Exception as e:
        print(f"⚠️ Routing error: {e}")
        return {"msg": state["msg"], "tool_result": None}

该函数的工作流程如下：
* 使用大型语言模型（LLM）分析用户请求。
* 根据可用工具列表（如 get_weather、web_search）决定是否调用工具。
* 如果决定调用工具，则提取所需参数并调用 call_mcp_tool 函数。
* 将工具调用的结果存入状态中。如果无需工具，则 tool_result 为 None。

2. 生成自然语言回答

（此部分内容将在后续小节详细展开，其职责是根据当前状态（包含原始消息和可能的工具调用结果）生成最终面向用户的自然语言回复。）

generate_response 函数负责生成最终的自然语言回复，确保无论是否使用了工具，AI 代理都能以流畅、对话式的方式响应用户。

# mcp_client.py
def generate_response(state: S):
    """Use OpenAI to generate a natural language response"""
    if state.get("tool_result") is None:
        # 未使用工具，直接由 LLM 生成回复
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[
                {"role": "system", "content": "You are a helpful assistant."},
                {"role": "user", "content": state["msg"]}
            ],
            temperature=0.7
        )
        return {
            "msg": state["msg"],
            "tool_result": state.get("tool_result"),
            "result": response.choices[0].message.content
        }

    # 格式化工具结果以供 LLM 使用
    tool_data = json.dumps(state["tool_result"], indent=2)

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a helpful assistant. Use the tool results provided to answer the user's question in a natural, conversational way. Be concise but informative."},
            {"role": "user", "content": f"User question: {state['msg']}nnTool results:n{tool_data}nnProvide a helpful answer based on this information. Please limit your answer in 100 words."}
        ],
        temperature=0.7
    )

    return {
        "msg": state["msg"],
        "tool_result": state["tool_result"],
        "result": response.choices[0].message.content
    }

构建 LangGraph 工作流

接下来，我们使用 LangGraph 将各个节点连接起来，形成一个完整的代理工作流。

# mcp_client.py
g = StateGraph(S)
g.add_node("route", route_request)
g.add_node("respond", generate_response)
g.set_entry_point("route")
g.add_edge("route", "respond")
g.add_edge("respond", END)
graph = g.compile()

该图定义了以下执行流程：
* 入口点：route 节点，负责判断用户请求是否需要调用工具。
* 核心节点：respond 节点，接收路由决策和可能的工具结果，生成最终的自然语言答案。
* 流程：route → respond → END。

工作流可视化

我们可以将构建的图结构保存为可视化图像，以便更直观地理解代理的执行逻辑。

# mcp_client.py
# ---------------- SAVE GRAPH VISUALIZATION ----------------
try:
    # 生成并保存图为 PNG 格式
    png_data = graph.get_graph().draw_mermaid_png()
    with open("langgraph_diagram.png", "wb") as f:
        f.write(png_data)
    print("📊 Graph visualization saved as 'langgraph_diagram.png'n")
except Exception as e:
    print(f"⚠️ Could not generate graph visualization: {e}")
    print("Note: Install graphviz system dependency if needed: brew install graphvizn")

LangGraph 工作流示意图

4. 测试 AI 代理

现在，让我们通过几个测试查询来验证代理的功能。

# mcp_client.py
tests = [
    "What's the weather in Mumbai?",
    "Search for latest news about AI",
]

for t in tests:
    result = graph.invoke({"msg": t})
    print(f"AGENT: {result['result']}n")

在终端中运行脚本：

python mcp_client.py

测试 1：查询天气

用户查询：What‘s the weather in Mumbai?

代理输出：

================================================================================
USER: What‘s the weather in Mumbai?
================================================================================
🤖 Routing decision: The user is asking for the current weather in Mumbai, which requires the get_weather tool.
AGENT: The current weather in Mumbai (Bombay) is 29°C (85°F) with smoky conditions. The humidity is at 33%, and there‘s a wind speed of 15 km/h. It feels like 32°C. If you‘re heading out, be mindful of the smoke and stay hydrated!

分析：代理正确识别了查询意图，调用了 get_weather 工具获取孟买的实时天气数据，并生成了整合工具结果的友好回复。

测试 2：搜索新闻

用户查询：Search for latest news about AI