SIMPLE MCP Service

SIMPLE平台客户端MCP服务，提供与SIMPLE平台的API交互能力。

快速

配置环境变量

创建 .env 文件：

# .env 文件内容
SIMPLE_ADDRESS=10.8.2.64:1100
SIMPLE_USERNAME=admin
SIMPLE_PASSWORD=your_password
SIMPLE_TIMEOUT=30

启动服务

uv run python -m simple.simple_mcp_server

服务将在 http://localhost:8001 启动。

AsyncSimpleClient 使用指南

AsyncSimpleClient 是 SIMPLE 平台的异步客户端，提供完整的 API 交互能力。

基本用法

import asyncio
from simple.simple_client import AsyncSimpleClient, format_datetime
from utils.logger import logger

async def main():
    # 创建客户端实例（自动从环境变量加载配置）
    async with AsyncSimpleClient() as client:
        # 使用客户端方法...
        models = await client.get_valid_models(factory_id=31)
        logger.info(f"Models result: {models}")

asyncio.run(main())

客户端初始化

# 方式1：使用环境变量配置
client = AsyncSimpleClient()

# 方式2：手动指定参数
client = AsyncSimpleClient(
    address="10.8.2.64:1100",
    username="admin",
    password="your_password",
    timeout=30
)

# 方式3：使用 async with 上下文管理器
async with AsyncSimpleClient() as client:
    # 自动管理连接生命周期
    pass

核心方法

1. 获取模型列表

async def get_valid_models(
    factory_id: int,           # 机组ID
    name_filter: str = "仿真回写"  # 模型名称过滤
) -> list

示例：

async with AsyncSimpleClient() as client:
    # 获取机组31的所有仿真回写模型
    models = await client.get_valid_models(factory_id=31)
    
    # 获取机组30的特定模型
    models = await client.get_valid_models(
        factory_id=30,
        name_filter="CCPP#2-YIXING-HCQ"
    )
    
    # 遍历模型
    for model in models:
        logger.info(f"ID: {model['id']}, Name: {model['mdName']}")

返回值结构：

[
    {
        "id": 123,
        "mdName": "仿真回写-机组A",
        "runtimeModelGuid": "abc123",
        "mdCategoryCode": "DIC_MODEL_CAT_EVENT"
    }
]

2. 创建回写任务

async def create_auto_writeback_task(
    task_name: str,              # 任务名称
    factory_id: int,             # 机组ID
    starttime: str,              # 开始时间 (ISO 8601格式)
    endtime: str,                # 结束时间 (ISO 8601格式)
    model_id: int,               # 模型ID
    model_guid: str,             # 模型GUID
    model_name: str,             # 模型名称
    model_category: str,         # 模型类别代码
    interval: int = 60,          # 回写间隔(秒)
    continue_on_error: bool = True  # 出错时是否继续
) -> dict

示例：

async with AsyncSimpleClient() as client:
    starttime = format_datetime("2025-01-01")
    endtime = format_datetime("2025-01-02")
    
    result = await client.create_auto_writeback_task(
        task_name="Dataintegration_仿真回写-机组A(2025-01-01_2025-01-02)",
        factory_id=31,
        starttime=starttime,
        endtime=endtime,
        model_id=123,
        model_guid="abc123",
        model_name="仿真回写-机组A",
        model_category="DIC_MODEL_CAT_EVENT",
        interval=60
    )
    
    task_id = result["data"]["id"]  # 获取创建的任务ID

3. 启动回写任务

async def start_auto_writeback_task(task_id: int) -> dict

示例：

async with AsyncSimpleClient() as client:
    result = await client.start_auto_writeback_task(task_id=12345)
    
    if result["code"] == 200:
        logger.info(f"任务启动成功: {result['data']['name']}")
    else:
        logger.error(f"任务启动失败: {result.get('message')}")

4. 删除回写任务

async def delete_auto_writeback_task(task_id: int) -> dict

示例：

async with AsyncSimpleClient() as client:
    result = await client.delete_auto_writeback_task(task_id=12345)
    logger.info(f"删除结果: {result}")

5. 批量执行回写任务

async def writeback_run(
    factory_id: int,
    starttime: str,
    endtime: str,
    name_filter: str = "仿真回写",
    isdelete: bool = False,      # 完成后是否删除任务
    delete_delay: int = 300      # 删除延迟(秒)
)

示例：

async with AsyncSimpleClient() as client:
    starttime = format_datetime("2025-01-01")
    endtime = format_datetime("2025-01-02")
    
    # 批量执行回写任务（不删除）
    await client.writeback_run(
        factory_id=31,
        starttime=starttime,
        endtime=endtime,
        name_filter="仿真回写"
    )
    
    # 批量执行并在完成后删除任务
    await client.writeback_run(
        factory_id=31,
        starttime=starttime,
        endtime=endtime,
        isdelete=True,
        delete_delay=300  # 等待5分钟后删除
    )

6. 日期格式化工具

def format_datetime(dt_str: str) -> str

示例：

from simple.simple_client import format_datetime

# 将日期字符串转换为ISO 8601格式
starttime = format_datetime("2025-01-01")
# 返回: "2025-01-01T15:57:58.745Z"

完整示例

import asyncio
from simple.simple_client import AsyncSimpleClient, format_datetime
from utils.logger import logger

async def complete_workflow():
    """完整的回写任务工作流示例"""
    async with AsyncSimpleClient() as client:
        # 1. 获取模型列表
        factory_id = 31
        models = await client.get_valid_models(factory_id, "仿真回写")
        
        if not models:
            logger.warning("未找到有效模型")
            return
        
        logger.info(f"找到 {len(models)} 个模型")
        
        # 2. 准备时间参数
        starttime = format_datetime("2025-01-01")
        endtime = format_datetime("2025-01-02")
        
        # 3. 为每个模型创建并启动任务
        for model in models:
            try:
                # 创建任务
                task_name = f"Dataintegration_{model['mdName']}({starttime}_{endtime})"
                create_result = await client.create_auto_writeback_task(
                    task_name=task_name,
                    factory_id=factory_id,
                    starttime=starttime,
                    endtime=endtime,
                    model_id=model["id"],
                    model_guid=model["runtimeModelGuid"],
                    model_name=model["mdName"],
                    model_category=model["mdCategoryCode"]
                )
                
                task_id = create_result["data"]["id"]
                logger.info(f"创建任务成功: {task_name} (ID: {task_id})")
                
                # 启动任务
                await asyncio.sleep(2)  # 避免请求过于频繁
                start_result = await client.start_auto_writeback_task(task_id)
                
                if start_result["code"] == 200:
                    logger.info(f"启动任务成功: {task_name}")
                else:
                    logger.error(f"启动任务失败: {task_name}")
                
            except Exception as e:
                logger.error(f"处理模型 {model.get('mdName', 'Unknown')} 时出错: {str(e)}")

asyncio.run(complete_workflow())

FastMCP 客户端使用

通过 FastMCP 协议调用服务：

import asyncio
from fastmcp import Client
from fastmcp.client.transports import SSETransport
from utils.logger import logger

async def main():
    transport = SSETransport(url="http://localhost:8001/sse")
    client = Client(transport=transport)
    
    async with client:
        # 获取工具列表
        tools = await client.list_tools()
        logger.info(f"可用工具: {[t.name for t in tools]}")
        
        # 调用工具
        result = await client.call_tool("get_valid_models", {"factory_id": 31})
        logger.info(f"工具调用结果: {result}")

asyncio.run(main())

环境变量配置

| 变量名 | 描述 | 默认值 | |--------|------|--------| | SIMPLE_ADDRESS | 服务器地址 | 10.8.2.64:1100 | | SIMPLE_USERNAME | 登录用户名 | admin | | SIMPLE_PASSWORD | | SIMPLE_TIMEOUT | 请求超时时间(秒) | 30 | | LOG_LEVEL | 日志级别 | INFO |

项目结构

mcp-simple/
├── simple/
│   ├── __init__.py
│   ├── simple_client.py      # AsyncSimpleClient 异步客户端
│   ├── simple_mcp_server.py  # FastMCP 服务实现
│   ├── llm_mcp_example.py    # LLM 调用示例
│   └── test_sse.py           # SSE 测试脚本
├── utils/
│   ├── __init__.py
│   ├── logger.py             # Loguru 日志配置
│   └── pkgs/
│       └── __init__.py       # 常用工具包
├── deploy/
│   ├── docker-compose.yml    # Docker Compose 配置
│   ├── Dockerfile            # Docker 镜像构建
│   ├── k3s-deployment.yaml   # K3s 部署配置
│   ├── Jenkinsfile           # Jenkins CI/CD
│   ├── prometheus/           # Prometheus 监控配置
│   └── scripts/              # 部署脚本
├── .env                      # 环境变量配置
├── pyproject.toml            # 项目配置
└── README.md                 # 项目文档

故障排除

常见问题

1. 连接失败

   • 检查服务器地址和端口是否正确
   • 确保网络能够访问目标服务器
   • 验证用户名和密码是否正确

2. 认证失败

   • 检查 SIMPLE_USERNAME 和 SIMPLE_PASSWORD 环境变量
   • 确认账户权限是否有效

3. 请求超时

   • 增大 SIMPLE_TIMEOUT 环境变量值
   • 检查服务器负载情况

日志查看

# 设置日志级别为 DEBUG
export LOG_LEVEL=DEBUG
uv run python -m simple.simple_mcp_server

许可证

MIT License