原创 阿凯 2025-05-26 18:31 上海
今年关MCP 的概念非常火,作为技术一线者,若把测试工具都改造成符合 MCP 服务协议标准,然后接入AI Agent,打造一个集万千工具于一体的智能管家来提效,这个设想就需要我们不断往下拆解成可落地的任务模块,这里我们先从造数开始。
💡作者设想将测试工具改造为符合MCP协议,并接入AI Agent,以实现智能测试数据生成的目的,从而提高测试效率。
🛠️文章介绍了基于FastAPI框架构建社区造数服务,并使用uv工具管理依赖,通过FastAPI-MCP库简化了MCP服务的接入流程。
✅文章详细阐述了FastAPI-MCP库的安装和使用方法,包括自定义配置、工具命名等,并提供了与Cursor等AI Agent工具的集成案例。
⚙️通过将FastAPI服务暴露为MCP工具,结合Cursor等AI Agent,用户只需用自然语言描述测试需求,即可自动完成复杂的造数任务,例如创建测试账号、发布动态等。
🚀文章强调了在接入AI Agent时,清晰的接口文档和规范的命名对AI理解和调用的重要性,并提供了调优案例。
原创 阿凯 2025-05-26 18:31 上海
今年关MCP 的概念非常火,作为技术一线者,若把测试工具都改造成符合 MCP 服务协议标准,然后接入AI Agent,打造一个集万千工具于一体的智能管家来提效,这个设想就需要我们不断往下拆解成可落地的任务模块,这里我们先从造数开始。
中间件相关配置全部通过 ARK 来管理,项目结构如下:## uv命令
1. 安装uv : `curl -LsSf https://astral.sh/uv/install.sh | sh`
2. 创建环境 - 自定义环境名称和Python版本 `uv venv tools_venv --python 3.12`
3. 激活环境 `source tools_venv/bin/activate`
4. 安装依赖包 `uv pip install -r pyproject.toml`
## 本地启动项目
直接运行main.py文件中的main方法即可,debug模式自己pycharm中设置
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
统一部署到公司的发布平台,通过 http://{造数服务域名}/tools/docs#/ ,地址可以访问目前社区所有的造数接口。同时也对接了造数工厂,可以直接去造数工厂使用。## 项目结构
```bash
├── main.py # 启动 APP 入口文件
├── README.md # 开发手册
├── Dockerfile # Docker 镜像文件
├── alembic # alembic 迁移 DB 自动生成的相关文件
│ ├── README
│ ├── .env.py
│ ├── script.py.mako
│ └── versions # 存放每次迁移的版本,可用于回滚 DB 版本
├── alembic.ini # alembic 配置文件
├── app
│ ├── __init__.py # 注册 app
│ ├── api # api 开发目录
│ ├── core # app 的全局配置
│ ├── crud # 每个 table 的增删改查操作
│ ├── db # db 配置
│ ├── models # 存放表结构
│ ├── schemas # pydantic 模型
│ └── utils # 工具类
├── .pre-commit-config.yaml # 配置 git commit 时自动检测工具
└── pyproject.toml # 依赖库管理
```
基于上述代码 demo,我们通过 uvicorn 启动服务,当然也可以单独启动 MCP 服务。控制台输出如下,代表启动成功,接下来我们就可以使用 MCP 客户端工具进行连接使用了,这里使用 Cursor 来做演示。看图标显示绿色,无报错说明连接成功,这里也能看到 demo 中的 get_user_info_tool 方法作为 MCP 工具暴露了出来。演示到这里,说明了该方案是可行的。因为本文重点讲解采用的新方案,此处就不再多介绍,感兴趣的可以去看官方文档。# server.py
from mcp.server.fastmcp import FastMCP
from tools.tools_set import get_user_info
import uvicorn
# Create an MCP server
mcp = FastMCP("Demo")
async def get_user_info_tool(mobile: str) -> Coroutine[Any, Any, Any]:
"""根据输入的手机号获取用户信息
Args:
mobile: 手机号
"""
info = get_user_info(mobile)
return info
if __name__ == "__main__":
"""Initialize and run the server"""
# mcp.run(transport="sse")
"""Start the FastAPI server with uvicorn"""
uvicorn.run(app, host="0.0.0.0", port=8003)
from fastapi import FastAPI
import uvicorn
from fastapi_mcp import FastApiMCP
# Create (or import) a FastAPI app
app = FastAPI()
# Create an MCP server based on this app
mcp = FastApiMCP(app)
# Mount the MCP server directly to your app
mcp.mount()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
※ Server metadataname:MCP 服务名description:对 MCP 服务的描述※ Tool and schema descriptions创建 MCP 服务器时,可以通过修改 describe_all_responses ,把所有可能的响应模式包含在工具描述中,或通过更改 describe_full_response_schema 把完整的 json 包含在工具描述中。class FastApiMCP:
"""
Create an MCP server from a FastAPI app.
"""
def __init__(
self,
fastapi: Annotated[
FastAPI,
Doc("The FastAPI application to create an MCP server from"),
],
name: Annotated[
Optional[str],
Doc("Name for the MCP server (defaults to app.title)"),
] = None,
description: Annotated[
Optional[str],
Doc("Description for the MCP server (defaults to app.description)"),
] = None,
describe_all_responses: Annotated[
bool,
Doc("Whether to include all possible response schemas in tool descriptions"),
] = False,
describe_full_response_schema: Annotated[
bool,
Doc("Whether to include full json schema for responses in tool descriptions"),
] = False,
http_client: Annotated[
Optional[httpx.AsyncClient],
Doc(
"""
Optional custom HTTP client to use for API calls to the FastAPI app.
Has to be an instance of `httpx.AsyncClient`.
"""
),
] = None,
include_operations: Annotated[
Optional[List[str]],
Doc("List of operation IDs to include as MCP tools. Cannot be used with exclude_operations."),
] = None,
exclude_operations: Annotated[
Optional[List[str]],
Doc("List of operation IDs to exclude from MCP tools. Cannot be used with include_operations."),
] = None,
include_tags: Annotated[
Optional[List[str]],
Doc("List of tags to include as MCP tools. Cannot be used with exclude_tags."),
] = None,
exclude_tags: Annotated[
Optional[List[str]],
Doc("List of tags to exclude from MCP tools. Cannot be used with include_tags."),
] = None,
auth_config: Annotated[
Optional[AuthConfig],
Doc("Configuration for MCP authentication"),
] = None,
):
...
※ Customizing Exposed Endpoints include_operations , 暴露 operation_id=XXX 的接口 exclude_operations , 排除 operation_id=XXX 的接口 include_tags , 暴露 tags=XXX 的接口 exclude_tags ,排除 tags=XXX 的接口from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
mcp = FastApiMCP(
app,
name="My API MCP",
description="Very cool MCP server",
describe_all_responses=True,
describe_full_response_schema=True
)
mcp.mount()
工具命名FastAPI 中的路由通过 operation_id 参数来作 MCP 工具名称,如果没有显示命名,框架会自动生成一个。此处经测试,如果不显示命名,自动生成的名字不仅会很奇怪,还会影响 AI 造数的准确性,所以这里最好作好规范,必须要显示命名。from fastapi import FastAPI
from fastapi_mcp import FastApiMCP
app = FastAPI()
# 案例1:include_operations
mcp = FastApiMCP(
app,
include_operations=["get_user", "create_user"]
)
# 案例2:exclude_operations
mcp = FastApiMCP(
app,
exclude_operations=["delete_user"]
)
# 案例3:include_tags
mcp = FastApiMCP(
app,
include_tags=["users", "public"]
)
#案例4:exclude_tags
mcp = FastApiMCP(
app,
exclude_tags=["admin", "internal"]
)
# 案例5:Combined
mcp = FastApiMCP(
app,
include_operations=["user_login"],
include_tags=["public"]
)
mcp.mount()
# Auto-generated operation_id (something like "read_user_users__user_id__get")
async def read_user(user_id: int):
return {"user_id": user_id}
# Explicit operation_id (tool will be named "get_user_info")
async def read_user(user_id: int):
return {"user_id": user_id}
步骤第一步:引入 fastapi-mcp 第二步:main.py 中添加 MCP 服务第三步:也是工作量最大的一步,将每个造数接口都做显示命名,并且做好文档注释,写的越清楚 AI 造数的准确率越高,需要对应编写造数场景测试,共同完成最后一步:启动服务 uvicorn.run('main:app', host='0.0.0.0', port=8023, reload=True, workers=2) ,无报错基本就没有问题了。再通过 MCP 客户端工具连接使用即可python = "^3.12"
fastapi = "0.115.12"
fastapi-mcp ="0.3.1"
mcp="1.7.0"
pydantic = "^2.11.0"
pydantic-settings = "^2.2.0"
第二步:点击右上角设置 icon,进入 Cursor Settings,选择 MCP第三步:这里可以看到,在刚才 mcp.json 中配置的 MCP工具均加载过来,打开开关,运行状态显示为绿色,无报错并说明了服务接入正常,接下来就可以正常使用 Cursor 中的 Agent 进行对话了{
"mcpServers": {
"fastapi-mcp": {
"url": "http://localhost:8022/mcp",
"description": "本地开发环境MCP服务配置"
},
"tools-mcp": {
"url": "http://localhost:8011/mcp",
"description": "本地开发环境MCP服务配置"
},
"demo-mcp": {
"url": "http://localhost:8001/sse",
"description": "本地开发环境MCP服务配置"
},
"tools-mcp-prod": {
"url": "http://XXXXXX/mcp",
"description": "线上"
}
}
}
把这个造数需求发送给 AI,发现报错了。我们去代码中看下为何返回了 false,原来是因为接口返回非 200,排查下来是因为 t1 环境测试账号造数默认填了 111,不需要再加 111,所以接口直接 500 了。这里 AI 犯了两个错误:因为默认手机号都是 11 位的,这里 AI 不知道只需要传 8 位就行。我没有输入具体的手机号,所以按照代码逻辑应该是支持自动随机生成的,但是 AI 也不知道这个逻辑,“自作主张”给我传入了一个手机号。※ 调优后通过排查我们已经明确知道 AI 犯了哪些错误,那么我们针对这些错误去调优即可。所谓的调优主要就是修改文档注释,可以前后对比下注释内容。async def c_create_account(
env: str = Body(..., description='环境'),
phonenumber: str = Body(..., description='手机号'),
pwd: str = Body(..., description='密码'),
usernum: str = Body(None, description='数量'),
) -> Any:
"""
创建测试账号,默认111开头
args
env: 环境,默认:t1
phonenumber: 手机号
pwd: 密码,默认:test123
usernum: 数量
"""
※ 最终效果通过这个案例可以看到,准确率依赖我们对造数接口的文档注释,所以在实际使用过程中,前期需要我们不断地去调优,才能达到我们想要的效果。当然随着后续迭代,可能可以用更优雅的方式完成这个工作,比如再引入静态代码分析工具,通过 AI 编程自动完成注释。"""
创建测试账号,默认111开头,不用填写111,只需要后面8位
不传手机号phonenumber,默认随机生成手机号
args
env: 环境,默认:t1
phonenumber: 手机号,非必填,不填自动生成
pwd: 密码,默认:test123
usernum: 数量
"""
AI辅助创作,多种专业模板,深度分析,高质量内容生成。从观点提取到深度思考,FishAI为您提供全方位的创作支持。新版本引入自定义参数,让您的创作更加个性化和精准。
鱼阅,AI 时代的下一个智能信息助手,助你摆脱信息焦虑