Understanding MCP Architecture: Single-Server vs Multi-Server Clients

了解 MCP 架构：单服务与多服务客户端

现代 AI 驱动的应用需要高效且可扩展的架构，以便与多个外部服务交互。Multi-Client Protocol（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（推理 + 行动）Agent，决定如何求解给定的数学表达式。
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（Server-Sent Events） 传输方式运行，以支持实时事件流传输。

多服务客户端

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

工作原理

1. 初始化多个服务：客户端通过 stdio 连接到 Math 服务，并通过 SSE（Server-Sent Events） 连接到 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 驱动应用，高效地与多个服务交互。🚀

---