模拟计算服务器

一个无状态的模型上下文协议(MCP)服务器,用于运行科学模拟,适用于集成到聊天应用、交互式仪表板和AI工具中。

  • by cnye36
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 cnye36/simulator-mcp-server

使用说明

模拟MCP服务器

用于运行科学模拟的无状态模型上下文协议(MCP)服务器。非常适合将模拟功能集成到聊天应用程序、交互式仪表板和人工智能工具中。

🎯 概述

此MCP服务器提供 无状态计算服务 它运行模拟并返回结果。它旨在集成到聊天应用程序中,其中:

  • ✅ 您的聊天应用程序处理身份验证和用户数据
  • ✅ MCP服务器执行计算
  • ✅ 结果以JSON格式返回,以便立即使用或存储

主要特点:

  • 🚀 无状态操作 -无身份验证,无用户数据存储
  • ⚡ 快速预览模式 -针对实时交互式UI进行了优化(~50-100ms)
  • 📊 JSON数据返回 -直接获取仿真数据,无需下载文件
  • 🔧 多种模型 -支持流行病学、物理学、金融学和定制模型
  • 📈 综合录井 -用于调试的完整请求/响应日志记录
  • 🌐 HTTP传输 -基于HTTP的标准MCP协议

📋 目录

🚀 快速开始

地方发展

# Clone the repository
git clone <repository-url>
cd simulation-mcp-server

# Install dependencies
pip install -r requirements.txt

# Run in stdio mode (for local development)
python server.py

码头工人

# Build and run with Docker Compose
docker compose up --build

# Health check
curl http://localhost:8000/health

示例请求

curl -X POST http://localhost:8000/mcp \
  -H "Content-Type: application/json" \
  -d '{
    "method": "tools/call",
    "params": {
      "name": "simulate_model",
      "arguments": {
        "domain": "epidemiology",
        "model_type": "SIR",
        "parameters": {"beta": 0.3, "gamma": 0.1},
        "initial_conditions": {"S": 0.99, "I": 0.01, "R": 0},
        "time_span": {
          "start": 0,
          "end": 160,
          "steps": 400,
          "preview_mode": true
        },
        "return_data": true,
        "save_artifacts": false
      }
    }
  }'

✨ 特性

核心能力

  • 多种仿真模型:目前支持SIR(流行病学),计划支持Lotka-Volterra、Logistic、Projectile和蒙特卡洛模型
  • 灵活的时间网格:具有预览模式的可配置时间跨度,可实现快速渲染
  • 多个求解器:RK45、RK23和DOP853集成方法
  • 敏感性分析:可选的单向灵敏度分析支持
  • 度量计算:自动计算关键指标(峰值、时间等)

性能优化

  • 预览模式:将数据点减少到100,响应时间约为50-100ms
  • 可选工件:跳过CSV/PNG生成,以获得更快的响应
  • JSON数据返回:直接获取仿真数据,无需文件I/O开销

生产就绪

  • 健康检查端点: /health 用于部署监控
  • 综合录井:具有可配置级别的完整请求/响应日志记录
  • 错误处理:带有堆栈跟踪的详细错误消息
  • Docker支持:已准备好使用Dockerfile和docker-compose.yml
  • 渲染部署:预配置 render.yaml 便于部署

📦 安装

需求

  • Python 3.12+
  • pip或uv

依赖项

pip install -r requirements.txt

关键依赖关系:

  • mcp[cli]>=1.2.0 -模型上下文协议SDK
  • pydantic>=2.7 -数据验证
  • numpy>=1.26 -数值计算
  • scipy>=1.11 -科学计算和ODE求解器
  • matplotlib>=3.8 -绘图生成(可选,用于工件)

环境变量

变量 默认值 描述
MCP_TRANSPORT stdio 运输方式(streamable-http 用于生产)
STORAGE_PATH ./storage 保存工件的目录
PORT 8000 服务器端口
FASTMCP_HOST 0.0.0.0 要绑定的主机(使用 0.0.0.0 用于外部访问)
LOG_LEVEL INFO 记录冗长(DEBUG, INFO, WARNING, ERROR)

💻 用法

基本模拟

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# Connect to server
async with stdio_client(StdioServerParameters(
    command="python",
    args=["server.py"]
)) as (read, write):
    async with ClientSession(read, write) as session:
        # Run simulation
        result = await session.call_tool(
            "simulate_model",
            arguments={
                "domain": "epidemiology",
                "model_type": "SIR",
                "parameters": {"beta": 0.3, "gamma": 0.1},
                "initial_conditions": {"S": 0.99, "I": 0.01, "R": 0},
                "time_span": {
                    "start": 0,
                    "end": 160,
                    "steps": 400
                },
                "return_data": True
            }
        )
        
        print(result.content[0].text)

交互式UI模式(快速预览)

# Optimized for real-time slider interactions
result = await session.call_tool(
    "simulate_model",
    arguments={
        # ... parameters ...
        "time_span": {
            "start": 0,
            "end": 160,
            "steps": 600,
            "preview_mode": True  # Limits to 100 points
        },
        "return_data": True,
        "save_artifacts": False  # Skip file generation
    }
)

高品质出口模式

# Full resolution with artifacts
result = await session.call_tool(
    "simulate_model",
    arguments={
        # ... parameters ...
        "time_span": {
            "start": 0,
            "end": 160,
            "steps": 1000,
            "preview_mode": False
        },
        "return_data": True,
        "save_artifacts": True  # Generate CSV and PNG files
    }
)

📚 API 参考

工具: simulate_model

使用指定参数运行模拟并返回结果。

输入参数

{
  domain: "epidemiology" | "physics" | "finance" | "custom",
  model_type: "SIR" | "LotkaVolterra" | "Logistic" | "Projectile" | "MonteCarlo",
  parameters: { [key: string]: number },  // Model-specific parameters
  initial_conditions: { [key: string]: number },  // Initial state values
  time_span: {
    start: number,      // Start time
    end: number,        // End time
    steps: number,      // Number of time points (>= 2)
    preview_mode?: boolean  // Limit to 100 points for speed
  },
  method?: "RK45" | "RK23" | "DOP853",  // ODE solver method (default: "RK45")
  sensitivity?: { [key: string]: number },  // Optional sensitivity analysis
  tags?: string[],  // Optional tags for organization
  return_data?: boolean,  // Return data points in response (default: true)
  save_artifacts?: boolean  // Save CSV/PNG files (default: false)
}

输出格式

{
  status: "success" | "error",
  message: string,
  summary?: string,  // Human-readable summary
  metrics: {
    [key: string]: number  // Calculated metrics (e.g., I_peak, t_peak)
  },
  columns: string[],  // Column names (e.g., ["t", "S", "I", "R"])
  data?: Array<{     // Time series data points
    t: number,
    [state_var: string]: number
  }>,
  artifacts: Array<{
    kind: "csv" | "plot" | "json",
    path: string
  }>
}

示例:SIR模型

输入:

{
  "domain": "epidemiology",
  "model_type": "SIR",
  "parameters": {
    "beta": 0.3,   // Transmission rate
    "gamma": 0.1   // Recovery rate
  },
  "initial_conditions": {
    "S": 0.99,     // Susceptible
    "I": 0.01,     // Infected
    "R": 0.0       // Recovered
  },
  "time_span": {
    "start": 0,
    "end": 160,
    "steps": 400,
    "preview_mode": false
  },
  "return_data": true,
  "save_artifacts": false
}

输出:

{
  "status": "success",
  "message": "Simulation completed",
  "summary": "Peak infection 0.3743 at t ≈ 33.89",
  "metrics": {
    "I_peak": 0.3743,
    "t_peak": 33.89
  },
  "columns": ["t", "S", "I", "R"],
  "data": [
    {"t": 0.0, "S": 0.99, "I": 0.01, "R": 0.0},
    {"t": 0.4, "S": 0.9896, "I": 0.0102, "R": 0.0002},
    // ... more data points
  ],
  "artifacts": []
}

健康检查端点

获取 /health

返回服务器状态和配置:

{
  "status": "healthy",
  "service": "simulation-mcp",
  "version": "0.1.0",
  "storage": "/tmp/storage",
  "solvers": 1
}

🚀 部署

Docker部署

# Build image
docker build -t simulation-mcp-server .

# Run container
docker run -p 8000:8000 \
  -e MCP_TRANSPORT=streamable-http \
  -e PORT=8000 \
  -e FASTMCP_HOST=0.0.0.0 \
  simulation-mcp-server

渲染部署

  1. 连接GitHub存储库 渲染
  2. 渲染自动检测 render.yaml 配置
  3. 设置环境变量 (如果需要):
    • MCP_TRANSPORT=streamable-http
    • PORT=8000
    • FASTMCP_HOST=0.0.0.0
  4. 部署 -渲染处理其余部分!

这 render.yaml 文件预先配置了:

  • 基于Docker的部署
  • 健康检查端点
  • 无状态操作(不需要持久磁盘)
  • 环境变量

其他平台

服务器可以部署到支持以下功能的任何平台:

  • Python 3.12+
  • HTTP/HTTPS端点
  • 环境变量配置

示例:Heroku、Railway、Fly.io、AWS ECS、Google Cloud Run等。

🏗️ 建筑

设计理念

此MCP服务器是 无状态 -它执行计算并返回结果。它不会:

  • ❌ 存储用户数据
  • ❌ 处理身份验证
  • ❌ 管理用户会话
  • ❌ 持续模拟结果

您的应用程序应处理:

  • ✅ 用户身份验证(例如,通过Supabase)
  • ✅ 聊天历史和上下文
  • ✅ 存储模拟结果(如果需要)
  • ✅ 用户特定数据管理

请求流

┌─────────────────────────────────┐
│   Your Chat App (Supabase)      │
│   - User authenticated          │
│   - Chat context stored         │
│   - User wants simulation       │
└──────────────┬──────────────────┘
               │ HTTP POST
               │ (No auth needed)
               ▼
┌─────────────────────────────────┐
│   MCP Server (Stateless)        │
│   - Receives parameters         │
│   - Runs simulation             │
│   - Returns JSON data           │
│   - No storage, no auth         │
└──────────────┬──────────────────┘
               │ JSON Response
               │ (data + metrics)
               ▼
┌─────────────────────────────────┐
│   Your Chat App                 │
│   - Receives results            │
│   - Stores in Supabase (optional)│
│   - Displays to user            │
│   - Manages user context        │
└─────────────────────────────────┘

存储策略

推荐方法:

  1. 使用以下命令调用MCP服务器 return_data: true 和 save_artifacts: false
  2. 直接接收JSON数据
  3. 如果需要,存储在您自己的数据库中(例如,Supabase)
  4. 如果需要导出,则在客户端生成文件

看 ARCHITECTURE.md 详细的集成模式和Supabase模式示例。

📖 文档

综合文档可在 docs/ 目录:

🔧 发展

项目结构

simulation-mcp-server/
├── server.py              # Main MCP server implementation
├── models/                # Simulation model implementations
│   ├── __init__.py
│   └── sir.py            # SIR epidemiology model
├── schemas/               # JSON schemas for validation
│   └── simulate_model_input.json
├── docs/                  # Documentation
├── Dockerfile             # Docker configuration
├── docker-compose.yml     # Docker Compose setup
├── render.yaml            # Render deployment config
├── requirements.txt       # Python dependencies
└── README.md             # This file

添加新模型

  1. 在中创建新的模型文件 models/ (例如。, models/lotka_volterra.py)
  2. 实现求解器功能:
    def simulate_lotka_volterra(
        parameters: Dict[str, float],
        y0: Dict[str, float],
        tspan: Tuple[float, float],
        steps: int,
        method: str = "RK45"
    ) -> Tuple[np.ndarray, np.ndarray, Dict[str, float]]:
        # Implementation
        return t, Y, metrics
    
  3. 注册 server.py:
    SOLVERS = {
        ("epidemiology", "SIR"): simulate_sir,
        ("epidemiology", "LotkaVolterra"): simulate_lotka_volterra,
        # ...
    }
    
  4. 更新JSON架构 schemas/simulate_model_input.json

运行测试

# Run server in stdio mode for testing
python server.py

# Test with curl (HTTP mode)
curl http://localhost:8000/health

日志记录

日志记录功能全面且可配置:

# Set log level
export LOG_LEVEL=DEBUG  # DEBUG, INFO, WARNING, ERROR

# Run server
python server.py

日志包括:

  • 请求参数
  • 执行时间
  • 数据形状和度量
  • 错误堆栈跟踪

📊 性能基准

模式 步骤 预览 工件 数据返回 响应时间
互动 600 ✅ 是 ❌ 否 ✅ 是 ~50-100ms
标准 600 ❌ 否 ❌ 否 ✅ 是 ~150-200ms
全质量 1000 ❌ 否 ✅ 是 ✅ 是 ~300-500ms
仅限文件 600 ❌ 否 ✅ 是 ❌ 否 ~250-400ms

在渲染启动器层(0.5 CPU,512 MB RAM)上测量

🤝 贡献

欢迎投稿!需要改进的地方:

  • 其他模拟模型(Lotka-Volterra、Logistic、射弹、蒙特卡洛)
  • 灵敏度分析增强
  • 性能优化
  • 文档改进

📝 许可证

[在此处添加您的许可证]

🆘 支持

对于问题、疑问或贡献:

  1. 检查 文档 目录
  2. 查看日志 LOG_LEVEL=DEBUG
  3. 检查 建筑.md 集成模式
  4. 看 docs/LOGGING_GUIDE.md 用于故障排除

🎉 状态

生产准备就绪! ✅

  • [x] 已配置HTTP传输
  • [x] 健康检查端点
  • [x] 以JSON格式返回的数据
  • [x] 性能预览模式
  • [x] 综合测井
  • [x] Docker容器化
  • [x] 渲染部署配置
  • [x] 文件完整
  • [x] 改进了错误处理

快乐模拟! 🚀