志刚
发布于 2025-07-23 / 13 阅读
0
0

使用Python在阿里云百炼创建基于UVX的MCP -高血压诊断

基于 uvx 高性能工具链,开发者可一键完成 Python 项目打包并发布至 PyPI,同时无缝接入阿里云百炼平台。相比传统部署方式,**uvx **支持零依赖运行和自动环境隔离,大幅简化 MCP 服务上线流程。通过 uvx 调用工具时,可直接执行脚本而无需全局安装,结合阿里云百炼的 Agent 能力,实现 AI 服务的快速集成与高效运维,显著降低环境配置复杂度

背景: 在大模型对于数值计算,比如在做高血压的诊断时,如果数字计算,逻辑计算的时候会出现部分错误,大概训练后的错误率在6%左右。为了彻底解决该类问题,需要在智能体流程中加入mcp的服务进行计算

mcp的搭建步骤和逻辑如下:

环境准备和项目搭建

1.1 命令行当中安装uv

什么是uv:uv是一个用Rust开发的超高性能Python包管理器,旨在替代传统的工具链(如pip、venv、pip-tools等)

确保你电脑安装了Python,这里推荐安装Python3.10

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

pip install uv

1.2.项目搭建

在您的 电脑上创建并打开任意一个文件夹,如果是Windows在对应创建的文件路径地址栏当中直接输入cmd即可打开命令行

1)创建项目文件(这个文件名称需要改成你自己的,否则后续打包上传pipy上的时候会冲突

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

uv init --package --python 3.10 gxy-bailian-mcp-server

 

2)请注意,上述操作要求Python版本与这里的一致,否则在阿里云百炼上可能会引发错误。

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

cd gxy-bailian-mcp-server

 

创建虚拟环境并安装所需的包。

在命令行中,通过“uv venv”创建虚拟环境,随后激活相应的虚拟环境,并安装所需的依赖项。

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

# 创建虚拟环境
uv venv
# 在MacOs/Linux当中激活虚拟环境
source .venv/bin/activate
# 在Windows当中激活虚拟环境
.venv\Scripts\activate
# 安装依赖库
uv add "mcp[cli]" httpx bs4 dotenv dashscope

 

1.3 在PyCharm中打开刚刚命令行创建的文件夹

打开后,您会发现左侧的文件结构已准备就绪。备注:如果没有安装PyCharm,也可以使VSCode打开

1.4 在src/gxy_bailian_mcp_server下面创建app.py,用于搭建MCP服务

接下来我们编写代码,我们将把阿里云百炼大模型封装为高血压诊断的MCP服务,代码如下:

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

# -*- coding: utf-8 -*-
from mcp.server.fastmcp import FastMCP
from pydantic import BaseModel, Field
import random
import logging
from typing import Literal, Optional
 
# 初始化日志和 MCP 服务
logger = logging.getLogger('mcp')
mcp = FastMCP('hypertension-diagnosis-service', log_level='INFO')
 
# 不同测量场景下的诊断阈值
DIAGNOSIS_THRESHOLDS = {
    "诊室测量": {"sbp": 140, "dbp": 90, "desc": "诊室血压"},
    "动态血压监测": {"sbp": 130, "dbp": 80, "desc": "24小时动态血压平均值"},
    "家庭自测": {"sbp": 135, "dbp": 85, "desc": "家庭自测血压"}
}
 
# 枚举类型
MeasurementContext = Literal["诊室测量", "动态血压监测", "家庭自测"]
BloodPressureLevel = Literal["血压偏低", "正常血压", "正常高值", "1级高血压", "2级高血压", "3级高血压"]
 
 
# 输入模型:支持中文字段 + 忽略多余字段
class HypertensionRequest(BaseModel):
    model_config = {
        "populate_by_name": True,  # 允许通过字段名(如 sbp)或别名(如 收缩压)赋值
        "extra": "ignore"          # 忽略智能体传的额外字段(如 并发症、年龄)
    }
 
    sbp: int = Field(
        ...,
        alias="收缩压",
        description="当前收缩压值,单位 mmHg",
        gt=0,
        le=300
    )
    dbp: int = Field(
        ...,
        alias="舒张压",
        description="当前舒张压值,单位 mmHg",
        gt=0,
        le=200
    )
    history_sbp: Optional[int] = Field(
        None,
        alias="既往最高收缩压",
        description="既往最高收缩压值,单位 mmHg",
        ge=0,
        le=300
    )
    history_dbp: Optional[int] = Field(
        None,
        alias="既往最高舒张压",
        description="既往最高舒张压值,单位 mmHg",
        ge=0,
        le=200
    )
    measurement_context: MeasurementContext = Field(
        default="诊室测量",
        alias="测量场景",
        description="测量场景:诊室测量/动态血压监测/家庭自测,默认为诊室测量"
    )
 
 
# 输出模型
class HypertensionResponse(BaseModel):
    level: BloodPressureLevel = Field(
        ...,
        description="血压临床分级结果"
    )
    diagnosis: bool = Field(
        ...,
        description="是否达到高血压诊断标准"
    )
    explanation: str = Field(
        ...,
        description="诊断与分级依据说明"
    )
    guideline: str = Field(
        ...,
        description="参考的临床指南名称"
    )
    service_version: str = Field(
        "1.0.0",
        description="服务版本号"
    )
 
 
# 获取血压分级(符合“或”判断逻辑)
def get_level(sbp: int, dbp: int, history_sbp: int = None, history_dbp: int = None) -> str:
    eval_sbp = history_sbp if history_sbp is not None else sbp
    eval_dbp = history_dbp if history_dbp is not None else dbp
 
    if eval_sbp >= 180 or eval_dbp >= 110:
        return "3级高血压"
    if eval_sbp >= 160 or eval_dbp >= 100:
        return "2级高血压"
    if eval_sbp >= 140 or eval_dbp >= 90:
        return "1级高血压"
    if eval_sbp >= 120 or eval_dbp >= 80:
        return "正常高值"
    if eval_sbp < 90 or eval_dbp < 60:
        return "血压偏低"
    return "正常血压"
 
 
# 获取诊断结果
def get_diagnosis(sbp: int, dbp: int, context_key: str) -> tuple[bool, str]:
    thresholds = DIAGNOSIS_THRESHOLDS.get(context_key)
    if not thresholds:
        raise ValueError(f"未知的测量环境: {context_key}")
 
    is_hypertensive = sbp >= thresholds['sbp'] or dbp >= thresholds['dbp']
    status = "达到" if is_hypertensive else "未达"
    reason = f"{thresholds['desc']}{status}诊断标准({thresholds['sbp']}/{thresholds['dbp']} mmHg)"
    return is_hypertensive, reason
 
 
# MCP 工具定义
@mcp.tool(
    name="hypertension_diagnosis",
    description=(
        "根据患者血压值和测量场景进行高血压分级与诊断。"
        "支持中文字段输入:如“收缩压”、“舒张压”、“测量场景”。"
        "若未提供测量场景,默认使用【诊室测量】。"
        "可忽略智能体传入的无关字段(如并发症、年龄)。"
        "适用于百炼智能体健康咨询、慢病管理等场景。"
    )
)
async def hypertension_diagnosis(
    request: HypertensionRequest
) -> HypertensionResponse:
    """
    高血压诊断主函数
    """
    # 获取测量场景(已通过 Pydantic 校验)
    context = request.measurement_context
 
    # 防御性 fallback(防止未来扩展问题)
    if context not in DIAGNOSIS_THRESHOLDS:
        logger.warning(
            f"不支持的测量场景 '{context}',将 fallback 到 '诊室测量'"
        )
        context = "诊室测量"
 
    # 诊断用血压值:优先使用既往最高
    eval_sbp = request.history_sbp if request.history_sbp is not None else request.sbp
    eval_dbp = request.history_dbp if request.history_dbp is not None else request.dbp
 
    # 执行诊断
    is_hypertensive, reason = get_diagnosis(eval_sbp, eval_dbp, context)
 
    # 获取分级
    level = get_level(
        request.sbp,
        request.dbp,
        request.history_sbp,
        request.history_dbp
    )
 
    # 构造解释文本
    history_info = ""
    if request.history_sbp is not None and request.history_dbp is not None:
        history_info = f"(基于既往最高血压 {request.history_sbp}/{request.history_dbp} mmHg)"
 
    explanation = f"当前血压 {request.sbp}/{request.dbp} mmHg{history_info}:{reason}"
 
    # 随机选择参考指南
    guideline = random.choice([
        "中国高血压防治指南(2024年修订版)",
        "中国老年高血压管理指南(2023版)"
    ])
 
    return HypertensionResponse(
        level=level,
        diagnosis=is_hypertensive,
        explanation=explanation,
        guideline=guideline,
        service_version="1.0.3"
    )
 
 
def run():
    """启动 MCP 服务"""
    logger.info("MCP 服务启动中...")
    logger.info("服务将通过 stdio 通信,等待智能体调用...")
    mcp.run(transport='stdio')
 
 
if __name__ == '__main__':
    run()

 

1.5 在__init__.py当中引入上述的app.py

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

from .app import run
def main() -> None:
    run()

 

1.6. 在vscode编辑器中使用cline 插件进行mcp本地服务测试

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

{
  "mcpServers": {
    "bailian-mcp-server": {
      "command": "/Users/liuzg/miniforge3/bin/uv",
      "args": [
        "--directory",
        "/Users/liuzg/work/ai/gxy-mcp-server/src/gxy_mcp_server",
        "run",
        "app.py"
      ],
      "env": {
        "API_KEY": "****替换为自己的在百炼上的apikey"
      }
    }
  }
}

 

其中,/Users/liuzg/work/ai/gxy-mcp-server/src 是之前创建项目的路径。

app.py 是先前编写 Python 代码的文件名称。

API_KEY 对应的是阿里云百炼的 API-KEY。

上述对应的 \ 符号已全部替换为 /。

2.2 运行测试

运行成功

3. 项目打包以后发布到pypi

3.1 对上述Python项目进行打包

在命令行当中运行

--javascripttypescriptshellbashsqljsonhtmlcssccppjavarubypythongorustmarkdown

uv build

 

3.2 将项目发布到pypi当中(发布后,如果直接在百炼上去部署的时候可能找不到,需要等待10-30分钟等待阿里云镜像生效)

首先我们需要去pypi的官网去注册账号。PyPI · The Python Package Index

创建对应的Token,备注在创建token的时候需要 2FA双因素认证 - TOTP 因为涉及第三方内容,请自己搜索解决。

在pypi上创建自己的api_token, 复制这个pypi 开头的token。

之后使用 uv publish --token pypi-xxxxx 命令将项目上传到 pypi 中。

uv publish --token pypi-xxxxx 

这样就上传到pypi 了,Client Challenge

全球的 Python 开发者都可以查看我们的代码了

 

4. 将MCP服务部署到阿里云百炼当中

4.1 打开阿里云百炼的自定义MCP服务

这个里面有个重点注意,需要在部署脚本里面的名字,和pypi上的服务名称保持一致

 

{
  "mcpServers": {
    "gxy-bailian-mcp-server": {
      "name": "gxy-bailian-mcp-server",
      "command": "uvx",
      "args": [
        "gxy-bailian-mcp-server"
      ],
      "env": {
        "API_KEY": "your apikey"
      }
    }
  }
}

 

填写完成后,点击部署按钮。部署成功后,再次打开该 MCP 服务

 

工具显示已成功,运行测试时也未发现问题。

4.2 如果部署完成后发现工具报错,可能是函数计算未正确获取到对应的依赖项。

我们进入上面自定义的MCP服务的函数节点中。

 

5. 在智能体当中引入上面搭建的MCP服务

5.1 创建智能体应用

 

此处一定要在提示词中说明你的mcp工具的入参格式,否则会造成模型对mcp的工具调用的格式乱掉,导致mcp服务调用失败

 

 

测试成功没有问题即可。


评论