MCP 实用指南 SSE 传输开发示例入门-服务器端开发

摘要：将外部数据源与 AI 模型集成通常需要复杂且耗时的自定义编码。模型上下文协议（MCP）通过提供一个标准化框架，简化了这一过程，实现了无缝交互。

将外部数据源与 AI 模型集成通常需要复杂且耗时的自定义编码。模型上下文协议（MCP）通过提供一个标准化框架，简化了这一过程，实现了无缝交互。

在本文中，我们将逐步介绍如何构建一个使用服务器发送事件（SSE）传输的 MCP 服务器和 MCP 客户端。

MCP 是一个通用接口，允许 AI 工具与内容存储库、商业平台和开发环境无缝交互。通过提供标准化框架，MCP 提高了 AI 应用的相关性和上下文感知能力。

它使开发人员能够：

标准化 AI 与各种工具和 API 的交互。在切换应用程序时保持上下文。提高 AI 驱动系统的模块化和可组合性。

在运行 MCP 服务器和客户端之前，我们需要安装所需的依赖项并设置环境变量。

1. 安装所需包

创建项目并运行以下命令以安装所需的依赖项。

（1）使用uv工具初始化一个新的Python项目

uv init mcp-search

（2）进入刚才创建的项目目录

cd mcp-search

（3）使用uv安装项目依赖

uv add "mcp[cli]" openai python-dotenv requests"mcp[cli]" – 用于 MCP 客户端与服务器通信。openai – 用于与 DeepSeek 模型交互的 API 客户端。python-dotenv – 用于管理环境变量。requests – HTTP请求库，用于调用外部API。

如果在安装过程中，出现 network timeout 异常，可以重试。

这些命令设置了一个基本的MCP项目。uv是一个较新的Python包管理工具，是pip的替代品，提供更快的包安装和管理功能。

使用命令 tree -a 在终端中显示目录结构：

tree -a

2. 设置 .env 文件

在项目目录中创建一个 .env 文件并添加你的 API 密钥：

SERPER_API_KEY=your_serper_api_key_hereDEEPSEEK_API_KEY=your_deepseek_api_key_here

这可以确保敏感凭据保持安全。

让我们开始创建一个提供以下两种功能的 MCP 服务器：

使用 Serper API 的网络搜索工具。一个基本的加法函数。

下面是MCP服务器端代码开发过程。首先，在项目中创建 mcp_server.py 文件。

1. 导入所需模块

导入必要的库和模块（FastMCP, requests, os, load_dotenv）加载环境变量（主要是搜索API的密钥）初始化 MCP 服务器实例# 导入 FastMCP 类，用于创建 MCP 服务器from mcp.server import FastMCP# 导入 requests 库，用于发送 HTTP 请求import requests# 导入 os 模块，用于访问环境变量import os# 导入 load_dotenv 函数，用于从 .env 文件加载环境变量from dotenv import load_dotenv# 加载 .env 文件中的环境变量（如 SERPER_API_KEY）load_dotenv# 创建 FastMCP 服务器实例# 参数说明：# - "Hello World"：服务器名称/描述# - port=9000：指定服务器运行的端口号mcp = FastMCP("Hello World", port=9000)

代码简要说明：

FastMCP: 初始化 MCP 服务器。

dotenv: 从 .env 文件加载 API 密钥。

2. 使用 Serper API 的网络搜索工具

在 MCP 中配置工具

在 MCP 中，每个用 @mcp.tool 装饰器包装的函数都被视为一个工具。这使得功能模块化变得非常容易。工具的描述（description）和输入模式（input schema）帮助大语言模型根据用户的查询选择合适的工具。

大语言模型会检查工具描述和输入模式，以匹配用户的查询。

例如，如果用户问：“5 加 10 是多少？”大语言模型会识别这是一个数学运算，并自动选择加法工具。

下面这段代码设置了Serper Google搜索API的参数并定义了一个搜索函数。以下是详细注释：

# 从环境变量获取Serper API密钥API_KEY = os.getenv("SERPER_API_KEY")# 设置Serper Google搜索API的URL端点API_URL = "https://google.serper.dev/search"# 使用mcp.tool装饰器将函数注册为MCP工具，使其可被AI模型调用@mcp.tooldef serper_search(query: str) -> dict: """使用 Serper API 进行搜索""" # 设置HTTP请求头，包含API密钥和内容类型 headers = { "X-API-KEY": API_KEY, # 添加API密钥到请求头 "Content-Type": "application/json" # 设置内容类型为JSON } # 构造搜索参数 params = { "q": query # 设置搜索查询词 } try: # 发送GET请求到Serper API，传递请求头和JSON格式的参数 response = requests.get(API_URL, headers=headers, json=params) # 检查响应状态，如果不是200系列会抛出异常 response.raise_for_status # 检查状态码是否为200（成功） if response.status_code == 200: # 解析JSON响应 result = response.json # 打印请求和结果信息（注意：这里有个格式化错误，应为{query}而不是{qquery}） print("请求：{qquery}，返回数据: ", result) # 返回搜索结果 return result else: # 如果状态码不是200，返回错误信息 return {"error": f"请求失败，状态码: {response.status_code}"} except requests.exceptions.RequestException as e: # 捕获请求异常（如网络错误、超时等） print(f"请求异常: {e}") # 返回错误信息 return {"error": str(e)}

接收用户查询并从 Serper API 获取搜索结果。

优雅地处理 API 错误并返回结构化结果。

3. 基本算术工具

下面这个简单函数，用于演示如何向 MCP 添加工具。

# 使用mcp.tool装饰器将函数注册为MCP工具# 这样AI模型可以通过MCP协议调用此函数@mcp.tooldef add(a: int, b: int) -> int: """计算两个数的和""" # 函数文档字符串，描述函数功能 # 打印调试信息，显示正在计算哪两个数字的和 print(f"计算 {a} + {b}") # 执行加法运算并返回结果 # 参数a和b都是整数类型，返回值也是整数类型 return a + b

这个函数是一个简单的示例工具，演示了如何通过MCP协议注册和使用基本数学运算功能。它与serper_search函数一样，都被注册为可由AI模型调用的工具函数。

4. 使用 SSE 传输运行 MCP 服务器

SSE 传输通过 HTTP POST 请求实现服务器到客户端的流式传输，用于客户端到服务器的通信。

if __name__ == "__main__": # 判断当前脚本是否作为主程序直接运行（不是被导入为模块） # 此条件语句确保以下代码块仅在脚本直接执行时才运行 # 打印启动消息，通知用户服务器正在启动及端口号 print("Starting MCP server on port 9000...") # 调用mcp实例的run方法启动服务器 # transport="sse" 参数指定使用服务器发送事件(Server-Sent Events)作为通信传输方式 # SSE是一种HTTP连接，允许服务器向客户端推送实时更新 # 这使得AI模型可以通过HTTP连接调用注册的工具函数(serper_search和add) mcp.run(transport="sse")

这段代码是程序的入口点部分，有以下几个关键功能：

if __name__ == "__main__": 语句用于判断当前脚本是否直接被执行（而不是被其他模块导入）。Python 解释器在执行脚本时，会给予其一个特殊变量 __name__，当脚本直接运行时，该值为 "__main__"；当脚本被导入时，值为模块名。打印提示信息，通知用户 MCP 服务器正在端口 9000 上启动。调用 mcp.run 方法启动 MCP 服务器

如下是完成的 mcp_server.py 文件结构图：