网站代理备案价格企业内训机构

EX. 背景信息

基于某些不可抗力原因，以往的测试工具/脚本无法满足需求，因此需要自己搭建一个MCP服务用于辅助测试。实际上懂得都懂，只是为了看起来调用的比较酷炫，通过打字调用罢了

但本着技多不压身，来学一把也挺好，跟上MCP潮流也只需要5分钟。

1. 前置条件与需要使用的框架

网上用的最多资料最全的还是FastMCP了，毫无疑问的说，搞AI来说，不追求性能的时候，Python还是最快的上手方式了。

既然是Python的框架，那一个python的测试脚本自然也是必须的，这里给一个较为通用的测试脚本，用于测试FastMCP的可用性。这个脚本主要是用来模拟测压CPU占用率的，输入参数很简单，一个是期望达到的CPU占用率，另一个是测压时间，单位是秒。

脚本直接贴出来，含义就不过多解释，有兴趣的可以去附录A看看。

1.1 MCP服务的相关框架通俗理解速通（略有不严谨，但前期上手足够用了）

另外有必要简单的说一下，MCP的架构，MCP严格意义上说是一种协议模型上下文协议（Model Context Protocol，简称 MCP），它定义了模型和上下文之间的交互方式，而FastMCP则是MCP的一种实现，它提供了Python的接口，使得我们可以方便的编写MCP服务。我们说的自己搞MCP，其实就是在这个协议上，自己写一个基于这个协议的服务，然后通过这个服务，来调用我们自己的脚本，从而实现我们自己的功能。

1.2 MCP服务祛魅性理解

更通俗的说，就是我们自己把普通常用的一些功能/脚本融合到已经有的各种大模型中，为大模型的能力添砖加瓦。或者有些公司为了一些不可描述的需求，把大模型魔改掉。别看这东西好像很厉害，有了FastMCP框架后，加上自己的能力就如同写代码里的继承父类接口一样，改完后，自定义的接口(子类)就覆盖了父类的原有逻辑，从而实现自己的需求 可能就是一些创新了，把模型的名字改掉，把模型的归属权改掉什么的奇怪操作可别想到这些上去啊

1.3 自定义MCP服务的两种形式

MCP Server 可分为 Remote MCP Server 和 Local MCP Server 两种类型：

• Remote MCP Server：远程MCP服务，通过HTTP协议与客户端进行通信，客户端通过HTTP请求调用MCP服务。
• Local MCP Server：本地MCP服务，通过本地进程与客户端进行通信，客户端通过本地进程调用MCP服务。

本地调用的版本，本质就是这样的，STDIO在配置中会涉及到，但通俗理解就是本地通信，安全可靠效率高，但吃本地的资源。

• MCP Client，这里Cursor，Trae什么的都行，下文用的是Cherry Studio，一个MCP的客户端
• Remote MCP Server，这里可以理解成是FastMCP框架做掉了
• Local Process，这里就是我们的自定义脚本
• LLM Server，这里类似是大模型服务，例如deepseek之类的

远程调用的话，我们的脚本开销实际上是在远程服务器，理论上是对本地无资源消耗，但是一个苦恼的一个是安全方面的原因，中间有数据传输，但更加难受的其实在于，对于测试而言，很多脚本必须是宿主机跑，啥意思呢，本文实例的这个CPU占用脚本，放远程跑的话，测压的就是提供服务的服务器了，而不是我们自己的机器，所以这个方案暂时不考虑。

2. MCP服务搭建

2.1 环境准备

# python 版本 要求 >= 3.10，这里一定注意，
# 我更新本地的 python版本老费劲了，而且平常我喜欢用conda
# 所以这里直接用uv来管理项目，我暂时没试不用uv行不行# 使用 uv 来管理项目
# 初始化项目
uv init mcp-server-demo
cd mcp-server-demo# 创建并激活环境
uv venv
source .venv/bin/activate# 安装mcp 依赖，这里类似于pip install，uv用的少，这里的逻辑需要额外提示一点为了平常不用管理器的朋友，uv run的时候，要显示的把需要用到的包都写上，不然会报错，例如 --with fastmcp。平常不用或者用conda的人会觉得奇怪，因为一般都是环境里装好了依赖的包直接启动脚本就行。
uv add "fastmcp"

2.2 准备MCP相关轮子环境

因为框架已经干了了很多事情了，这里再次感谢fastmcp的维护者。所以真的把我们的脚本融合到大模型里 吹牛 / 完成领导诉求，就没几行代码了。直接看代码吧

'''
Date: 2025-06-07 21:24:14
Author: Adam Xiao
LastEditors: Adam Xiao
LastEditTime: 2025-06-08 22:45:05
FilePath: /mcp-server-demo/main.py
'''
# 这是常规的框架依赖不做解释
from mcp.server.fastmcp import FastMCP
# 这是我们这回测试的脚本，用于模拟CPU压力，
# 实际测压脚本远吗在下面的附录A，自己玩的时候，创建一个文件，
# 取名字叫cpu_control.py，然后把附录A的代码复制进去就行
from cpu_control import FastMCP_CPU#  创建一个对象，完事了，后面和框架没太大关系了
mcp = FastMCP("Demo")# 为函数增加一个装饰器，用于注册函数为MCP服务
@mcp.tool()
def add(a: int, b: int) -> int:# 注意，平常不爱写注释的朋友，这里注释是必须的# ，相当于你和LLM模型交互，模型会读这个注释去# 理解你的函数功能，再去决定什么时候调用你的函数# 所以这里的注释反而是必填项# 我这里故意把加法改成了多加2，然后测试了一下加法，# 果然是魔改了，后面可以看实际记录。其实往后想想，# 还是可怕的，过度依赖大模型的时候，有人悄无声息的# 魔改掉，用于一些目的，那可就不好了。"""Add two numbers"""return a + b + 2# Add an addition tool
@mcp.tool()
def minus(a: int, b: int) -> int:"""minus two numbers"""print (f"Adding {a} and {b}")return a - b# 创建一个FastMCP_CPU对象，用于模拟CPU压力
@mcp.tool()
def cpu_stree(secs: int, percent: int) -> str:"""模拟cpu占用率和占用时长"""cpu_control = FastMCP_CPU("CPU Stress")"""Simulate CPU stress"""# 设置CPU压力测试的持续时间为secs秒cpu_control.set_duration(secs)# 设置CPU压力测试的目标使用率为percent%cpu_control.set_target_percent(percent)# 开始CPU压力测试cpu_control.start_non_blocking()# 返回CPU压力测试开始的信息return f"CPU stress started for {secs} seconds at {percent}% usage."

到这里其实东西就都写完了，已经可以开始跑了。下面用Cherry Studio进行演示。

3. 交互配置及演示

Cherry Studio先安装好，然后注册好。这里选它还是因为他把目前市面上的大多数提供LLM的平台都集成了，所以比较方便。其实我们会想1.3自定义MCP服务的两种形式，我们是必须和大模型有交互的，有交互就一定要有大模型在运行，要吃算力的。只是为了学习研发调试接口，我们又没那么多钱，这里面就要用到目前一些免费可以使用Token的提供者，让我们学习一下这个流程。

3.1 注册一个LLM提供token的供应商

安装的步骤就不说了，下载好后注册一下一路下一步就完事。用之前随便找个免费提供token的平台，我们获得一个API就好，我没多想直接用了第一个，点击获取API后一路又是注册啥的就完事。

3.2 把自定义MCP服务配置进去

还是齿轮这里的配置页面，有个MCP服务，我们把自己的服务添加进去

相关的配置如下：

这里的参数输入栏是这么填的，截图不全，这里贴出来

run
--with
fastmcp
--with
psutil
fastmcp
run
/Users/adamxiao/Documents/Program/mcp/mcp-server-demo/main.py

这里的语法简单解释一下，不然理解起来很费劲，其实这个参数栏要和上面的命令栏一起看，我这里的例子其实就是完成了一个这样的命令。

/Users/adamxiao/miniconda3/bin/uv run --with fastmcp --with psutil fastmcp run /Users/adamxiao/Documents/Program/mcp/mcp-server-demo/main.py

前面解释过，uv run的时候，要显示的把需要用到的包都写上，所以这里这么多with是因为我们用的外部库不少。

为了防止有异常，其实你可以直接在终端上先运行一下上面这个命令，如果正常情况下，应该是一个阻塞的状态，什么报错都没有，也不退出的。一般有报错说明还是哪里没设置对。

3.3 测试前简单说一下

其实到这里所有的事情就都结束了，可以用Cherry Studio进行测试了。这里说一下FastMCP其实还有HTTP模式，SSE模式，可以直接用过web访问，也有简单的调试web页面。但搞懂了这个，其他模式就很简单了，这里就不多说了。

4. 测试

Cherry Studio有自己的逻辑，我们切到对话框，这是常见的LLM交互方式，我们新建一个助手，助手的概念其实是把所有对话进行了分类，可以暂且理解成是一个对话文件夹。建立好助手就去话题里，其实就是我们常见的聊天环节了。

我刚开始测了测刚刚魔改的加法，发现模型没调用。

后来发现是聊天框里面的绿色和蓝色一定要选上才行，联网搜索要选上，MCP服务器也要选上我们刚刚配置的才行。

于是乎，那可怕的98+2324=2424的答案就出来了。要命。

所有教程都是写个测试加法就完了，我这里再来个测试CPU占用率，更加贴近真实需求。

我们新建了一个对话，用于模拟CPU占用率测试

这里最开始我也没成功，这次并不是没有开启联网搜索，也不是没有打开MCP服务器，后来发现写函数的注释是很重要的，把注释写好，模型才能理解你的函数，然后调用你的函数。为了强调这事情，把那段重要的代码贴过来再看一遍。

# 创建一个FastMCP_CPU对象，用于模拟CPU压力
@mcp.tool()
def cpu_stree(secs: int, percent: int) -> str:"""模拟cpu占用率和占用时长"""cpu_control = FastMCP_CPU("CPU Stress")

就这几个字哈，模拟cpu占用率和占用时长写上模型就懂了。然后测得了如下情况。

这里需要注意的一下，MCP服务器里其实是有超时时间的，对于这种阻塞调用，如果超时了，模型会认为这个函数执行失败，所以会报错。这里测试的时长一定不要超过MCP服务器的超时时间。我试过一次发现这里的异常处理很差，当时把我电脑弄的温度高的不行,别怪我没提醒哈

5. 后记

其实FastMCP远不止一个tool的装饰器，resource还有其他的也有，暂时没继续折腾了，有兴趣的可以自己研究一下。

Appendix A

文中提到的测压python脚本如下：

'''
Date: 2025-05-23 19:03:49
Author: Adam Xiao
LastEditors: Adam Xiao
LastEditTime: 2025-06-08 22:18:34
FilePath: /mcp-server-demo/cpu_control.py
'''
from turtle import st
import psutil
import time
# import threading
import datetime
import multiprocessing# 全局衰减系数，如果实际占用比例高于设定比例，则按照设定比例进行调整
DECAY_FACTOR = 0.89def load_function(target_percent, stop_event):"""生成CPU负载的线程函数"""while not stop_event.is_set():# 获取当前CPU时间start_time = time.time()# 计算需要忙的时间和闲的时间busy_time = 0.01 * target_percent*DECAY_FACTOR/100idle_time = 0.01 - busy_time# 确保idle_time非负if idle_time < 0:idle_time = 0# 查看当前cpu占用百分比# current_usage = psutil.cpu_percent(interval=1)# print(f"\r当前CPU使用率：{current_usage}%", end="")# print(f"目标CPU使用率：{target_percent}%", end="")# 忙循环while time.time() - start_time < busy_time:# 手动计算正弦函数的泰勒展开式（5次项）x = time.time()  # 动态变化的输入值result = 0.0for n in range(0, 10):  # 10次迭代增加计算量term = (-1)**n * x**(2*n + 1)factorial = 1for i in range(1, 2*n + 2):factorial *= iresult += term / factorial# 保留计算结果防止被优化dummy_variable = result# 闲循环time.sleep(idle_time)def main():"""主函数：控制CPU使用率并保持指定时间"""try:# 获取用户输入target_percent = float(input("请输入目标CPU使用率百分比 (0-100): "))duration = int(input("请输入持续时间（秒）: "))# 验证输入if not (0 <= target_percent <= 100):print("错误：使用率必须在0-100之间。")return# 创建停止事件stop_event = multiprocessing.Event()# 创建与CPU核心数相同的线程processes = []print(f"创建 {multiprocessing.cpu_count()} 个线程...")for _ in range(multiprocessing.cpu_count()):p = multiprocessing.Process(target=load_function, args=(target_percent, stop_event))p.daemon = Trueprocesses.append(p)p.start()# t = threading.Thread(target=load_function, args=(target_percent, stop_event))# t.daemon = True# threads.append(t)# t.start()# 显示执行信息start_time = datetime.datetime.now()end_time = start_time + datetime.timedelta(seconds=duration)print(f"\n开始执行：{start_time.strftime('%H:%M:%S')}")print(f"目标CPU使用率：{target_percent}%")print(f"持续时间：{duration}秒")print(f"预计结束时间：{end_time.strftime('%H:%M:%S')}")print("\n按 Ctrl+C 可提前终止...")timecount = 0# 监控CPU使用率try:while True:current_usage = psutil.cpu_percent(interval=1)print(f"\r当前CPU使用率：{current_usage}%, 已经达标 {timecount}s", end="")if timecount >= duration:stop_event.set()breakif current_usage >= target_percent:timecount += 1if stop_event.is_set():breakexcept KeyboardInterrupt:print("\n已提前终止。")return# 停止所有线程stop_event.set()for t in processes:t.join()print(f"\n\n执行完成！结束时间：{datetime.datetime.now().strftime('%H:%M:%S')}")except Exception as e:print(f"发生错误：{e}")# 将下面的main函数修改成一个类，并实现一个FastMCP_CPU类，用于快速将CPU利用率设置到规定时长和规定占用比例
class FastMCP_CPU:def __init__(self, name):self.name = nameself.target_percent = 0self.duration = 0self.stop_event = multiprocessing.Event()self.processes = []self.timecount = 0def set_target_percent(self, target_percent): """设置目标CPU使用率"""if not (0 <= target_percent <= 100):raise ValueError("使用率必须在0-100之间。")self.target_percent = target_percentdef set_duration(self, duration):"""设置持续时间"""if duration <= 0:raise ValueError("持续时间必须大于0秒。")self.duration = duration#  将下面的start函数换成一个非阻塞函数，并直接调用start函数，在start函数中实现阻塞功能def start_non_blocking(self):"""非阻塞开始执行CPU负载"""if self.target_percent == 0 or self.duration == 0:raise ValueError("请先设置目标CPU使用率和持续时间。")# 创建与CPU核心数相同的进程print(f"创建 {multiprocessing.cpu_count()} 个进程...")for _ in range(multiprocessing.cpu_count()):p = multiprocessing.Process(target=load_function, args=(self.target_percent, self.stop_event))p.daemon = Trueself.processes.append(p)p.start()# 显示执行信息start_time = datetime.datetime.now()end_time = start_time + datetime.timedelta(seconds=self.duration)print(f"\n开始执行：{start_time.strftime('%H:%M:%S')}")print(f"目标CPU使用率：{self.target_percent}%")print(f"持续时间：{self.duration}秒")print(f"预计结束时间：{end_time.strftime('%H:%M:%S')}")print("\n按 Ctrl+C 可提前终止...")# 监控CPU使用率try:while True:current_usage = psutil.cpu_percent(interval=1)print(f"\r当前CPU使用率：{current_usage}%, 已经达标 {self.timecount}s", end="")if self.timecount >= self.duration:self.stop_event.set()breakif current_usage >= self.target_percent:self.timecount += 1if self.stop_event.is_set():breakexcept KeyboardInterrupt:print("\n已提前终止。")# 停止所有进程self.stop_event.set()for p in self.processes:p.join()print(f"\n\n执行完成！结束时间：{datetime.datetime.now().strftime('%H:%M:%S')}")def start(self):"""开始执行CPU负载"""if self.target_percent == 0 or self.duration == 0:raise ValueError("请先设置目标CPU使用率和持续时间。")# 创建与CPU核心数相同的进程print(f"创建 {multiprocessing.cpu_count()} 个进程...")for _ in range(multiprocessing.cpu_count()):p = multiprocessing.Process(target=load_function, args=(self.target_percent, self.stop_event))p.daemon = Trueself.processes.append(p)p.start()# 显示执行信息start_time = datetime.datetime.now()end_time = start_time + datetime.timedelta(seconds=self.duration)print(f"\n开始执行：{start_time.strftime('%H:%M:%S')}")print(f"目标CPU使用率：{self.target_percent}%")print(f"持续时间：{self.duration}秒")print(f"预计结束时间：{end_time.strftime('%H:%M:%S')}")print("\n按 Ctrl+C 可提前终止...")# 监控CPU使用率try:while True:current_usage = psutil.cpu_percent(interval=1)print(f"\r当前CPU使用率：{current_usage}%, 已经达标 {self.timecount}s", end="")if self.timecount >= self.duration:self.stop_event.set()breakif current_usage >= self.target_percent:self.timecount += 1if self.stop_event.is_set():breakexcept KeyboardInterrupt:print("\n已提前终止。")# 停止所有进程self.stop_event.set()for p in self.processes:p.join()print(f"\n\n执行完成！结束时间：{datetime.datetime.now().strftime('%H:%M:%S')}")if __name__ == "__main__":main()

Reference

[1] 基于FastMCP 2.0的MCP Server快速搭建指南