ai4mats-mcp-tools/tools/app_deploy_mcp_server.py

import os
import json
import httpx
from mcp.server.fastmcp import FastMCP

# ============================================
# 配置
# ============================================
def load_config():
    """加载配置"""
    config_path = os.path.join(os.path.dirname(__file__), "config.json")
    default_config = {
        "API_BASE_URL": "http://172.20.32.121:31213"
    }

    try:
        if os.path.exists(config_path):
            with open(config_path, "r", encoding="utf-8") as f:
                default_config.update(json.load(f))
    except Exception:
        pass

    return default_config


config = load_config()
API_BASE_URL = config["API_BASE_URL"]

# ============================================
# MCP 服务定义
# ============================================
mcp = FastMCP("智算应用部署")


# ============================================
# 通用请求封装
# ============================================
async def post_api(token: str, path: str, payload: dict) -> str:
    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json"
    }

    try:
        async with httpx.AsyncClient(timeout=60) as client:
            r = await client.post(
                f"{API_BASE_URL}{path}",
                json=payload,
                headers=headers
            )
            r.raise_for_status()
            return json.dumps(r.json(), indent=2, ensure_ascii=False)

    except httpx.HTTPStatusError as e:
        return f"❌ 接口错误 ({e.response.status_code})"
    except httpx.TimeoutException:
        return "❌ 请求超时"
    except Exception as e:
        return f"❌ 调用失败：{e}"


# ============================================
# 工具：创建应用
# ============================================
@mcp.tool()
async def 创建应用(
    token: str,
    应用名称: str,
    服务类型: str = "机器学习",
    来源: int = 2,
    描述: str = "",
    图标URL: str = "",
    标签: str = ""
) -> str:
    return await post_api(token, "/api/mmp/service", {
        "service_name": 应用名称,
        "service_type": 服务类型,
        "source": 来源,
        "description": 描述 or 应用名称,
        "img_url": 图标URL,
        "tag": 标签 or 应用名称
    })


# ============================================
# 工具：创建应用版本
# ============================================
@mcp.tool()
async def 创建应用版本(
    token: str,
    应用ID: int,
    版本号: str,
    版本描述: str = "",
    资源类型: str = "GPU",
    镜像资源ID: int = 362,
    镜像ID: int = 59,
    启动命令: str = "test_pl_vor.py",
    代码仓库名称: str = "钙钛晶体特征提取代码",
    Git地址: str = "https://openi.pcl.ac.cn/somunslotus/somun202508191526161/tree/master",
    Git分支: str = "master",
    模型ID: int = 5,
    模型名称: str = "原子参杂识别模型1",
    模型版本: str = "v11",
    模型路径: str = "fanshuai/model/75/fanshuai_model_20260202150153/v11/model",
    部署类型: str = "web"
) -> str:

    payload = {
        "service_id": 应用ID,
        "version": 版本号,
        "description": 版本描述 or 版本号,
        "resource_type": 资源类型,
        "image_resource": {"id": 镜像资源ID},
        "image": {"imageID": 镜像ID},
        "command": 启动命令,
        "code_config": {
            "code_repo_name": 代码仓库名称,
            "git_url": Git地址,
            "git_branch": Git分支
        },
        "model": {
            "id": 模型ID,
            "name": 模型名称,
            "version": 模型版本,
            "path": 模型路径
        },
        "source": 2,
        "deploy_type": 部署类型
    }

    return await post_api(token, "/api/mmp/service/version", payload)


# ============================================
# 工具：重启应用
# ============================================
@mcp.tool()
async def 重启应用(token: str, 应用ID: int, 版本ID: int) -> str:
    return await post_api(token, "/api/mmp/service/version/restart", {
        "service_id": 应用ID,
        "version_id": 版本ID
    })


# ============================================
# 工具：停止应用
# ============================================
@mcp.tool()
async def 停止应用(token: str, 应用ID: int, 版本ID: int) -> str:
    return await post_api(token, f"/api/mmp/service/version/stop/{应用ID}/{版本ID}", {})


# ============================================
# 工具：删除应用版本
# ============================================
@mcp.tool()
async def 删除应用版本(token: str, 应用ID: int, 版本ID: int) -> str:
    return await post_api(token, f"/api/mmp/service/version/{应用ID}/{版本ID}", {})


# ============================================
# 工具：查询应用状态
# ============================================
@mcp.tool()
async def 查询应用状态(
    token: str,
    应用ID: int,
    页码: int = 0,
    每页数量: int = 10
) -> str:
    url = (
        f"/api/mmp/service/version/list"
        f"?page={页码}&size={每页数量}"
        f"&service_id={应用ID}&source=2"
    )
    return await post_api(token, url, {})


# ============================================
# ⚠️ 注意：不写 mcp.run()
# 由主网关统一启动
# ============================================