精通MCP服务器与客户端从入门到精通第一篇

精通 MCP Server｜Client：从入门到SSE实战

掌握MCP Server与Client的核心技术，紧跟大厂实践，迎接tool call的未来潮流。本教程将深度解析MCP的核心优势，并手把手指导你构建基于SSE（Server-Sent Events）模式的Server和Client，彻底摆脱传统stdio模式的本地限制，实现远程、动态可插拔的tool call。

为什么用 MCP？

MCP与传统tool call相比，最显著的特点在于其动态可插拔特性：

• 动态性：工具组（tool group）中可调用的tool是动态变化的。当你在某个tool group server中新增、删除或修改tool时，所有已连接的client都能自动感知到这些变化。这在生产环境中极为实用，无需重启客户端即可生效。
• 通用性与分层架构：MCP清晰划分了Resource、Tools、Root等概念，架构更加标准化。
• 生态与大厂认可：除了OpenAI公开兼容，阿里、腾讯等国内大厂也已全面支持MCP。这意味着tool call正经历一场重大变革——从原始的「刀耕火种」时代迈进「青铜器」时代。

SSE Transport（传输模式）

官方示例基于stdio模式（快速入门server），这是最早期的一种模式：Server和Client必须运行在同一台机器上。Client启动时需要指定本地脚本路径：

uv run client.py path/to/server.py  # python server

Client 代码示例：

server_params = StdioServerParameters(
  command="python",
  args=[server_script_path],
  env=None
)

stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))

但如果你需要管理大量server，将它们全部集中在一台机器上并频繁重启，就会失去动态可插拔的优势。因此MCP还支持SSE（Server-Sent Events）模式，这是一种基于网络的远程调用方式。

官方配置中指定传输类型：

def run(self, transport: Literal["stdio", "sse"] = "stdio")
....

官方解释：

SSE传输通过HTTP POST请求实现服务器到客户端的流式传输，从而实现客户端到服务器的通信。在以下场景中推荐使用SSE：

• 仅需要服务器到客户端的流式传输
• 使用受限网络
• 实现简单更新

由于基于网络工作，官方特别给出了安全注意事项：

安全警告：DNS重新绑定攻击
如果未妥善保护，SSE传输可能容易遭受DNS重新绑定攻击。为防止此类攻击，请遵循以下措施：

1. 始终验证传入SSE连接的Origin标头，确保其来源符合预期
2. 在本地运行时，避免将服务器绑定到所有网络接口（0.0.0.0）——仅绑定到本地主机（127.0.0.1）
3. 为所有SSE连接实施适当的身份验证

若缺乏这些保护，攻击者可能通过DNS重新绑定从远程网站与本地MCP服务器进行交互。

构建 SSE Server

官方SDK提供了mcp.server.fastmcp.FastMCP用于快速构建Server（也可以使用mcp.server.Server，但FastMCP更为便捷）。下面我们将构建一个简单的Server，包含两个经典tool：天气查询和货币汇率转换。

1. 初始化 FastMCP server

from mcp.server.fastmcp import FastMCP

# 默认的host是0.0.0.0，port是8000
mcp = FastMCP("sse_weather", host="localhost", port=9990)

更多可配置参数请参考mcp.server.fastmcp.server.Settings类。

2. 编写 Tool（类似 LangChain 的装饰器方式）

@mcp.tool()
async def get_weather(city: str) -> str:
    """Get weather information for a city.

    Args:
        city: Name of the city
    """

    return f"{city} is sunny, enjoy it!"

@mcp.tool()
def convert(amount: float, currency_from: str, currency_to: str) -> float:
    """use latest exchange rate to convert currency

    Args:
      amount: the amount of currency to convert
      currency_from: the currency to convert from
      currency_to: the currency to convert to
    """
    return amount * 0.8

3. 运行 Server

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

本文使用uv（astral.sh/uv）作为Python包管理器。uv使用非常便捷，带有浓厚的Rust包管理风格（因为它基于Rust开发）。最新版本的PyCharm也已支持uv作为包管理器。

将上述代码保存为sse_weather.py，然后启动：

uv run sse_weather.py

如果看到以下输出，表示成功：

INFO:     Started server process [24973]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://localhost:9990 (Press CTRL+C to quit)

构建 SSE Client

1. 创建 Client 类

import asyncio
from contextlib import AsyncExitStack
from typing import Optional

from mcp import ClientSession, ListToolsResult
from mcp.client.sse import sse_client
from mcp.types import TextContent, ImageContent, EmbeddedResource, Tool, CallToolResult


class MCPClient:
    def __init__(self):
        # Initialize session and client objects
        self.session: Optional[ClientSession] = None
        self.exit_stack = AsyncExitStack()

2. 添加连接 Server 的方法

    async def connect_to_server(self, server_url: str):
        """Connect to an MCP server

        Args:
            server_url: URL of the server
        """

        sc = sse_client(url=server_url)
        sse_transport = await self.exit_stack.enter_async_context(sc)
        sse, write = sse_transport
        cs = ClientSession(sse, write)
        self.session = await self.exit_stack.enter_async_context(cs)

        await self.session.initialize()

3. 添加 list_tools 和 tool_call 方法

    async def list_tools(self) -> list[Tool]:
        """List a vailable tools"""
        response: ListToolsResult = await self.session.list_tools()
        print(response.model_dump())
        return response.tools


    async def tool_call(self, tool_name: str, args: dict) -> list[TextContent | ImageContent | EmbeddedResource]:
        """Call a tool by name with arguments"""
        response: CallToolResult = await self.session.call_tool(tool_name, args)
        return response.content

4. 资源清理方法

    async def cleanup(self):
        """Clean up resources"""
        await self.exit_stack.aclose()

5. 测试主函数

async def main():
    client = MCPClient()
    try:
        await client.connect_to_server("http://127.0.0.1:9990/sse")
        await client.list_tools()
        r = await client.tool_call("get_weather", {"city": "Beijing"})
        print(r)

    finally:
        await client.cleanup()

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

使用 uv 运行 Client：

uv run sse_client.py

运行结果展示

ListToolsResult 输出

Client 成功连接到 Server 后，首先打印出工具列表（JSON 格式）：

{
  "meta": "None",
"nextCursor": "None",
"tools": [
    {
      "name": "get_weather",
      "description": "Get weather information for a city.\n\n    Args:\n        city: Name of the city\n    ",
      "inputSchema": {
        "properties": {
          "city": {
            "title": "City",
            "type": "string"
          }
        },
        "required": [
          "city"
        ],
        "title": "get_weatherArguments",
        "type": "object"
      }
    },
    {
      "name": "convert",
      "description": "use latest exchange rate to convert currency\n\n    Args:\n      amount: the amount of currency to convert\n      currency_from: the currency to convert from\n      currency_to: the currency to convert to\n    ",
      "inputSchema": {
        "properties": {
          "amount": {
            "title": "Amount",
            "type": "number"
          },
          "currency_from": {
            "title": "Currency From",
            "type": "string"
          },
          "currency_to": {
            "title": "Currency To",
            "type": "string"
          }
        },
        "required": [
          "amount",
          "currency_from",
          "currency_to"
        ],
        "title": "convertArguments",
        "type": "object"
      }
    }
  ]
}

调用 tool 的结果

调用get_weather后，Client 打印返回内容：

[TextContent(type='text', text='Beijing is sunny, enjoy it!', annotations=None)]

至此，一个完整的SSE模式下Server和Client的交互演示已顺利完成。

你是一名 AI 行业编辑，请围绕下面这条热点输出一份资讯解读：
热点：精通MCP服务器与客户端从入门到精通第一篇要求：
1. 先用一句话解释这条热点在讲什么
2. 再总结它为什么重要
3. 说明会影响哪些 AI 产品或内容方向
4. 最后给出 3 个适合资讯站使用的标题