精通 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重新绑定攻击。为防止此类攻击，请遵循以下措施：
- 始终验证传入SSE连接的Origin标头，确保其来源符合预期
- 在本地运行时，避免将服务器绑定到所有网络接口（0.0.0.0）——仅绑定到本地主机（127.0.0.1）
- 为所有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的交互演示已顺利完成。