理解 MCP 架构：单服务器客户端 vs 多服务器客户端

现代 AI 驱动的应用程序需要高效且可扩展的架构来与多个外部服务交互。多客户端协议 (MCP) 提供了一个强大的框架来与各种工具服务器通信，从而实现 AI 模型和外部计算之间的无缝交互。在本博客中，我们将通过比较单服务器和多服务器客户端来探讨 MCP 架构的优势。

数学服务器 (math_server.py)

数学服务器充当 MCP 工具服务器，提供基本的数学运算，如加法和乘法。它监听传入的请求，处理工具调用，并返回结果。

代码分解

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Math")

@mcp.tool()
def add(a: int, b: int) -> int:
    """Add two numbers"""
    return a + b

@mcp.tool()
def multiply(a: int, b: int) -> int:
    """Multiply two numbers"""
    return a * b

if __name__ == "__main__":
    mcp.run(transport="stdio")

• 定义一个名为 Math 的 MCP 服务器。
• 注册 add(a, b) 和 multiply(a, b) 作为可用工具。
• 使用 stdio 传输运行，允许通过标准输入/输出进行通信。

单服务器客户端

单服务器客户端连接到一个 MCP 服务器，该服务器公开一组可在计算中使用的工具。让我们分析一下 single_server_client.py 脚本，该脚本与 math_server.py 交互以执行数学运算。

工作原理

1. 启动数学服务器：math_server.py 作为 MCP 服务器运行，公开两个工具：add(a, b) 和 multiply(a, b)。
2. 初始化 Stdio 连接：客户端创建一个子进程，通过 stdio 传输与数学服务器通信。
3. 加载可用工具：客户端从服务器获取可用工具（add、multiply）。
4. 调用 AI 模型：AI 模型 GPT-4o 使用 LangGraph 的 ReAct (推理 + 行动) 代理来决定如何解决给定的数学表达式。
5. 执行计算：模型生成工具调用，首先调用 add(3,5)，然后调用 multiply(8,12)，最后返回计算出的答案。

代码分解 (single_server_client.py)

server_params = StdioServerParameters(
    command="python",
    args=["server/math_server.py"],
)

async with stdio_client(server_params) as (read, write):
    async with ClientSession(read, write) as session:
        await session.initialize()
        tools = await load_mcp_tools(session)
        agent = create_react_agent(model, tools)
        events = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})

输出打印：

----------------------------------------------------------------------------
The result is 96

MCP 通信流程（序列图）

sequenceDiagram
    participant Client as Client Script
    participant LangGraph as LangGraph React Agent
    participant ClientSession as MCP ClientSession
    participant stdio as stdio Transport Layer
    participant MathServer as Math Server Process
    participant Tools as Math Tools (add/multiply)
    participant LLM as GPT-4o Model

    Note over Client: Starts execution with asyncio.run()
    Client->>MathServer: Launch as subprocess with stdio parameters
    
    Client->>stdio: Create stdio client connection
    stdio-->>Client: Return read/write streams
    
    Client->>ClientSession: Initialize session
    ClientSession->>stdio: Send initialize message
    stdio->>MathServer: Forward initialize request
    MathServer-->>stdio: Return server capabilities
    stdio-->>ClientSession: Return initialization response
    ClientSession-->>Client: Session initialized
    
    Client->>ClientSession: load_mcp_tools()
    ClientSession->>stdio: Request available tools
    stdio->>MathServer: Forward tools request
    MathServer->>Tools: Discover registered tools (add, multiply)
    Tools-->>MathServer: Return tool descriptions
    MathServer-->>stdio: Return tool descriptions
    stdio-->>ClientSession: Return tool descriptions
    ClientSession-->>Client: Return langchain-compatible tools
    
    Client->>LangGraph: Create React agent with model and tools
    LangGraph-->>Client: Return agent instance
    
    Client->>LangGraph: agent.ainvoke("what's (3 + 5) x 12?")
    LangGraph->>LLM: Generate reasoning and tool calls
    
    Note over LLM: Decides to use add(3, 5) first
    LLM-->>LangGraph: Return reasoning and add(3, 5) tool call
    
    LangGraph->>ClientSession: Call add(3, 5)
    ClientSession->>stdio: Send tool execution request
    stdio->>MathServer: Forward tool execution request
    MathServer->>Tools: Execute add(3, 5)
    Tools-->>MathServer: Return result (8)
    MathServer-->>stdio: Return tool execution result
    stdio-->>ClientSession: Return tool execution result
    ClientSession-->>LangGraph: Return add result (8)
    
    LangGraph->>LLM: Generate next reasoning step with add result
    
    Note over LLM: Decides to use multiply(8, 12) next
    LLM-->>LangGraph: Return reasoning and multiply(8, 12) tool call
    
    LangGraph->>ClientSession: Call multiply(8, 12)
    ClientSession->>stdio: Send tool execution request
    stdio->>MathServer: Forward tool execution request
    MathServer->>Tools: Execute multiply(8, 12)
    Tools-->>MathServer: Return result (96)
    MathServer-->>stdio: Return tool execution result
    stdio-->>ClientSession: Return tool execution result
    ClientSession-->>LangGraph: Return multiply result (96)
    
    LangGraph->>LLM: Generate final answer with all results
    LLM-->>LangGraph: Return final answer "The result is 96"
    LangGraph-->>Client: Return complete event trace and answer
    
    Client->>Client: Print final answer

天气服务器 (weather_server.py)

天气服务器是另一个 MCP 工具服务器，提供实时天气信息。它监听传入的天气查询，并返回最新的天气数据。

启动服务器的命令：

python server/weather_server.py

代码分解

from mcp.server.fastmcp import FastMCP

mcp = FastMCP("Weather")

@mcp.tool()
async def get_weather(location: str) -> str:
    """Get weather for location."""
    return "It's always sunny in New York"

if __name__ == "__main__":
    mcp.run(transport="sse")

• 定义一个名为 Weather 的 MCP 服务器。
• 注册 get_weather(location) 作为可用工具。
• 使用 SSE (服务器发送事件) 传输运行，允许实时事件流。

多服务器客户端

多服务器客户端可以同时与多个 MCP 服务器交互。multi_server_client.py 脚本通过与 math_server.py 和 weather_server.py 通信来演示这一点。

工作原理

1. 初始化多个服务器：客户端使用 stdio 连接到 Math 服务器，并通过 SSE (服务器发送事件) 连接到 Weather 服务器。
2. 从多个服务器加载工具：AI 模型可以访问来自两个服务器的工具。
3. 并行处理：客户端现在可以在同一执行中调用来自不同服务器的工具。
4. AI 驱动的响应：模型使用多个 MCP 服务器处理数学表达式和天气查询。

代码分解 (multi_server_client.py)

async with MultiServerMCPClient(
    {
        "math": {
            "command": "python",
            "args": ["server/math_server.py"],
            "transport": "stdio",
        },
        "weather": {
            "url": "http://localhost:8000/sse",
            "transport": "sse",
        },
    }
) as client:
    agent = create_react_agent(model, client.get_tools())
    math_response = await agent.ainvoke({"messages": "what's (3 + 5) x 12?"})
    weather_response = await agent.ainvoke({"messages": "what is the weather in nyc?"})

多服务器客户端响应

来自多服务器客户端的输出如下所示：

{
  "math_response": "The result of (3 + 5) x 12 is 96.",
  "weather_response": "The weather in NYC is currently sunny."
}

MCP 架构的优势

• 可扩展性：易于扩展到多个服务。
• 并行执行：允许并发查询不同的服务器。
• 灵活的通信：支持多种传输层（stdio、sse 等）。
• AI 集成：将 AI 模型与外部计算服务无缝连接。

代码设置和执行

要克隆和执行代码，请按照以下步骤操作：

git clone https://github.com/commitbyrajat/langgraph-mcp-integration.git
cd langgraph-mcp-integration
rye sync  # Sync dependencies
rye run python server/weather_server.py &
rye run python client/single_server_client.py
rye run python client/multi_server_client.py

MCP 架构使开发人员能够构建强大的 AI 驱动的应用程序，这些应用程序可以有效地与多个服务交互。🚀