社区造数服务接入MCP|得物技术

发布于:2025-05-29 ⋅ 阅读:(21) ⋅ 点赞:(0)

一、背景

今年 MCP 的概念非常火,市面上也涌现出了一大批 MCP 相关工具。作为技术一线者,都会按捺不住地去实操一下,很早的时候就有个设想,如果把我们的测试工具都改造为符合 MCP 服务协议标准,然后全部接入 AI Agent,打造一个集万千工具于一体的智能管家来帮助我们提效,是不是一个很完美的设想。很多宏伟或者天马行空的想法想要真正的落地,必然需要不断向下,拆解成可落地的任务模块,这里我们先从造数开始。

​二、AI 造数设想

在实际业务需求测试中,我们依赖的测试数据需要很多前置的数据要求,这时候会涉及到分步使用不同的造数脚本。比如团长拉新做任务,需要一个 30 天内没发过动态的账号,加入团队,发一篇动态,动态过一审,过二审,阅读数满足 300 个。

为了完成这个场景的造数,我们需要去造数工厂、接口自动化、脚本代码等平台找对应的造数工具,分别去执行才能完成这一系列的操作。可以从下图中看到,总计需要 6 个步骤才能完成。如果不是熟悉所有的业务,哪怕有现成的造数脚本,组合起来使用还是有一定的门槛。

那么在 AI 风行的年代,我们想要实现的是:按照用户输入的测试数据要求,能够按照已有造数能力自动编排,生成对应的测试数据给用户使用。

最终实现效果案例:我需要一个团长拉新的测试数据,要求是 30 天内没有发过动态,进入团队 A,然后发布一条动态,需要过一审风控审核,二审标注,最后需要获得 300 个阅读数。

AI 造数自动去造数池子中寻找对应的造数接口,按照提问的顺序要求来依次执行造数,最后返回给用户对应的测试账号。

这里不再重复介绍 MCP 的概念,我们参考官方给出的 client-server 通用架构图来画一个 AI 造数的架构图,便于理解在落地到 AI 造数的场景,我们可以做哪些事。本篇文章主要就讲解了图中的其中一环,落地社区造数服务的 MCP 接入。

三、社区造数服务tools

框架介绍

社区的造数服务技术栈是基于 FastAPI 框架实现的,通过 uv工具来管理依赖库、虚拟环境等,这个工具亲测的确比传统的 pip 或者 poetry 等工具更好用。从安装 uv到启动项目,只要 4 步就能无痛搞定环境,不用担心本地其他环境的干扰。

## 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)

中间件相关配置全部通过 ARK 来管理,项目结构如下:

## 项目结构


```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  # 依赖库管理
```

统一部署到公司的发布平台,通过 http://{造数服务域名}/tools/docs#/ ,地址可以访问目前社区所有的造数接口。同时也对接了造数工厂,可以直接去造数工厂使用。

改造思路

老方案-基于 MCP Python SDK

早在出现 MCP 这个概念的时候,我就想过有天把我们的造数服务通过 MCP 工具暴露出来,这样就可以非常方便的集成各种 Agent,打造 AI 造数。在出现这个 FastAPI-MCP 框架之前,想要把造数服务改造成支持 MCP ,就需要通过引入 MCP 依赖库来实现。但这个方案对于已有的造数服务来说改造成本有些高,可以看老方案的案例。

从官方文档面向服务器开发者 - MCP 中文文档中可以找到有对应的 MCP Python SDK,主要就是安装 MCP 这个依赖库。这里举一个简单的 demo,通过手机号查询用户信息的方法。可以很清晰的看出来这个 SDK 的语法结构是需要 @mcp.tool()  这个装饰器来修饰,那么原有的造数服务暴露出来的所有接口方法是否都需要改造,这仍有一定的成本(未考虑其他复杂场景情况下)。

# 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")


@mcp.tool()
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)

基于上述代码 demo,我们通过 uvicorn 启动服务,当然也可以单独启动 MCP 服务。控制台输出如下,代表启动成功,接下来我们就可以使用 MCP 客户端工具进行连接使用了,这里使用 Cursor 来做演示。

看图标显示绿色,无报错说明连接成功,这里也能看到 demo 中的 get_user_info_tool 方法作为 MCP 工具暴露了出来。演示到这里,说明了该方案是可行的。因为本文重点讲解采用的新方案,此处就不再多介绍,感兴趣的可以去看官方文档。

四、FastAPI-MCP

安装运行

“Expose your FastAPI endpoints as Model Context Protocol (MCP) tools, with Auth! ”

这是引用官网介绍的第一句话,翻译过来大概的意思就是:把你的 FastAPI 服务作为 MCP 工具暴露出来成为现实!

1.安装 FastAPI-MCP 库

  uv add fastapi-mcp  or  uv pip install fastapi-mcp 

2.使用 FastAPI-MCP,只需要 3 行代码就能把 FastAPI 框架改造成一个 MCP 服务

3.通过 uvicorn 启动服务器,使用http://localhost:8000/mcp 来访问 MCP server

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)

用法介绍

自定义配置

通过看源码 FastApi-MCP 类,基本能清晰的看出来各个参数的用处,这里将介绍几个常用的。

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,
    ):
    ...

※ Server metadata

name:MCP 服务名

description:对 MCP 服务的描述

※ Tool and schema descriptions

创建 MCP 服务器时,可以通过修改 describe_all_responses ,把所有可能的响应模式包含在工具描述中,或通过更改 describe_full_response_schema 把完整的 json 包含在工具描述中。

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()

※ Customizing Exposed Endpoints

 include_operations , 暴露 operation_id=XXX 的接口

 exclude_operations , 排除 operation_id=XXX 的接口

 include_tags , 暴露 tags=XXX 的接口

 exclude_tags ,排除 tags=XXX 的接口

组合使用:

 include_operations 和 exclude_operations 不能同时使用

 include_tags 和 exclude_tags 不能同时使用

 include_operations 和 include_tags 可以组合使用,匹配任一个条件就满足

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()

工具命名

FastAPI 中的路由通过 operation_id 参数来作 MCP 工具名称,如果没有显示命名,框架会自动生成一个。此处经测试,如果不显示命名,自动生成的名字不仅会很奇怪,还会影响 AI 造数的准确性,所以这里最好作好规范,必须要显示命名。

# Auto-generated operation_id (something like "read_user_users__user_id__get")
@app.get("/users/{user_id}")
async def read_user(user_id: int):
    return {"user_id": user_id}


# Explicit operation_id (tool will be named "get_user_info")
@app.get("/users/{user_id}", operation_id="get_user_info")
async def read_user(user_id: int):
    return {"user_id": user_id}

​五、接入造数服务

框架升级及改造

在接入的时候,要查一下官方文档要求的 Python,FastAPI 等版本,先进行框架升级,防止出现不兼容的问题。这项通过管理工具安装依赖库时能自动校验,其他一些兼容问题在启动服务后根据实际场景一一去解决即可。这里推荐使用 uv 工具进行管理,亲测比之前的 poetry 更好用。

列几个核心库的版本,都是验证过没有兼容问题的。在过程中也是遇到一些兼容问题花了点时间,因为 FastAPI-MCP 框架比较新,网上资料还不全,遇到没法解决的问题大家可以去项目 issue 中找,提升解决问题效率。

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"

步骤

第一步:引入 fastapi-mcp 

第二步:main.py 中添加 MCP 服务

第三步:也是工作量最大的一步,将每个造数接口都做显示命名,并且做好文档注释,写的越清楚 AI 造数的准确率越高,需要对应编写造数场景测试,共同完成

最后一步:启动服务 uvicorn.run('main:app', host='0.0.0.0', port=8023, reload=True, workers=2) ,无报错基本就没有问题了。再通过 MCP 客户端工具连接使用即可

接入 Cursor

改造完之后的造数服务成功对外暴露了 MCP 服务,现在我们可以通过 MCP 客户端去连接使用了,这里选用了 Cursor,因为 Cursor 使用的人比较多,同时集成了市面上的主流大模型。

步骤

第一步:创建一个 mcp.json,按照标准 json 配置即可

{
  "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": "线上"
    }
}
}

第二步:点击右上角设置 icon,进入 Cursor Settings,选择 MCP

第三步:这里可以看到,在刚才 mcp.json 中配置的 MCP工具均加载过来,打开开关,运行状态显示为绿色,无报错并说明了服务接入正常,接下来就可以正常使用 Cursor 中的 Agent 进行对话了

实操演练

我们现在只希望使用造数能力,因此我们可以指定刚才配置的 MCP 工具。

场景化案例

需求:给手机号为 11120210001 的用户发布一个点评动态,并且通过风控一审。

这里注意一下提问方式,因为我们没有对大模型进行特别的训练,AI 不一定知道 111 开头的是我们测试使用的虚拟手机号,有可能会误解为 userId,所以我们需要告诉 AI 这是一个手机号。

可以看到在这个 demo 中, Agent 自动帮我们分了三步去调用对应的 MCP tool,第一步通过我们输入的手机号去获取 userId,第二步通过 userId 去发布点评动态,第三步通过点评动态 id 去通过风控一审。原本需要三步完成的造数场景,现在通过一句话描述就完成了。


调优案例

需求:随机创建 10 个测试账号

※  调优之前

造数代码,主要看文档注释内容。

@router.post('/create-account', operation_id="create_account",summary="创建测试账号")
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,发现报错了。我们去代码中看下为何返回了 false,原来是因为接口返回非 200,排查下来是因为 t1 环境测试账号造数默认填了 111,不需要再加 111,所以接口直接 500 了。

这里 AI 犯了两个错误:

1.因为默认手机号都是 11 位的,这里 AI 不知道只需要传 8 位就行。

2.我没有输入具体的手机号,所以按照代码逻辑应该是支持自动随机生成的,但是 AI 也不知道这个逻辑,“自作主张”给我传入了一个手机号。

※  调优后

通过排查我们已经明确知道 AI 犯了哪些错误,那么我们针对这些错误去调优即可。所谓的调优主要就是修改文档注释,可以前后对比下注释内容。

"""
创建测试账号,默认111开头,不用填写111,只需要后面8位
不传手机号phonenumber,默认随机生成手机号


args
    env: 环境,默认:t1
    phonenumber: 手机号,非必填,不填自动生成
    pwd: 密码,默认:test123
    usernum: 数量
"""

※  最终效果

通过这个案例可以看到,准确率依赖我们对造数接口的文档注释,所以在实际使用过程中,前期需要我们不断地去调优,才能达到我们想要的效果。

当然随着后续迭代,可能可以用更优雅的方式完成这个工作,比如再引入静态代码分析工具,通过 AI 编程自动完成注释。

​六、总结

技术实践成果

通过将社区造数服务改造成符合 MCP(Model Context Protocol) 标准的工具,我们成功实现了以下目标:

AI 驱动的测试数据自动化

用户通过自然语言描述需求,AI Agent 可自动编排造数接口生成复杂测试数据,将原本需手动执行 3 步的操作简化为一步指令。

低成本框架升级

基于 fastapi-mcp 框架,仅需少量代码改造即可将 FastAPI 服务快速接入 MCP 协议,解决了传统 SDK 方案的高适配成本问题。

工具链整合

通过对接 Cursor 等 AI 工具平台,验证了 MCP 协议在跨平台协作中的可行性,为后续构建“社区智能管家”奠定技术基础。

核心实践经验

注释即规范

AI 调用接口的准确性高度依赖代码注释的清晰度。通过优化接口文档(如参数默认值、输入格式说明),可显著提升 Agent 的任务解析成功率。

渐进式调优 

初期需通过人工干预优化 Agent 的接口调用逻辑,未来可引入代码静态分析工具自动生成标准化注释。

未来优化方向

动态编排增强 

当前接口调用为线性执行,后续可探索基于依赖关系的动态编排(如并行执行独立步骤、自动重试失败操作)。

多 Agent 协作

结合领域知识库与测试断言工具,实现从“造数”到“验证”的全链路 AI 自治。

协议扩展性

探索 MCP 与更多协议(如 OpenAPI)的互操作性,提升工具服务的跨平台复用能力。

价值与启示

本次实践印证了 “AI+协议化工具” 在测试领域的巨大潜力:降低技术门槛 (非技术人员可直接描述需求)、提升执行效率 (分钟级操作秒级完成)、释放创新空间 (复杂场景的自动化长链路测试)。

随着 MCP 生态的完善,测试工程将逐步从“工具堆砌”走向“智能协作”,为研发效能带来质的突破。

​七、感想

“我不是英雄,只是一个拿锤子的约德尔人”

站在巨人的肩膀上总是能看的更高更远,追随技术大牛们的步伐,把 AI 应用到工作中、生活中。回想九年前初入测试行业时捧读的《Google 软件测试之道》,书中“人类智慧的最后一英尺”已然越来越近。重读了 2022 年在公司内部博客发表的《Google 软件测试之道:结合实践的总结》一文,发现仅仅过了3 年,如果现在再去写,又是完全不一样的想法了,技术的发展已发生翻天覆地的变化。

此刻回望测试领域的演进曲线,愈发感到:「拿锤者」的价值不在于挥舞工具的姿态,而在于持续校准认知坐标的能力 。当 AI 重构测试链路的每个环节时,唯以「锤者」的务实与「巨人」的视野双轨并行,方能在技术洪流中锚定价值支点。

加油吧!不忘初心,你我终将能抵达一个又一个“终点”!

往期回顾

1.CSS闯关指南:从手写地狱到“类”积木之旅|得物技术

2.从零实现模块级代码影响面分析方案|得物技术

3.以细节诠释专业,用成长定义价值——对话@孟同学 |得物技术

4.得物可观测平台架构升级:基于GreptimeDB的全新监控体系实践

5.得物自研DGraph4.0推荐核心引擎升级之路

文 / 阿凯

关注得物技术,每周更新技术干货

要是觉得文章对你有帮助的话,欢迎评论转发点赞~

未经得物技术许可严禁转载,否则依法追究法律责任。


网站公告

今日签到

点亮在社区的每一天
去签到