MCP Scenario Engine

MCP场景引擎

用于确定性仿真和场景规划的MCP服务器

概述

MCP场景引擎是一个模型上下文协议（MCP）服务器，它提供了一个模拟空间，人工智能代理可以在其中查询系统状态、执行操作，并以可追溯的方式模拟影响。

支持复杂配方现在，您可以通过JSON定义的规则对复杂的系统进行建模，包括博弈论、经济学、健康跟踪和科学模拟。

用例

• 🎯 项目规划：资源分配、时间线模拟
• 🏗️ 基础设施：容量规划、负载测试
• 🔬 科学建模：物理、化学、生物模拟
• 📊 经济与金融：市场动态、拍卖、投资
• 🎮 博弈论：战略互动、纳什均衡、进化动力学
• 🏋️ 健康与健身：身体成分、营养、训练优化
• 🌳 假设分析：备选方案的时间线分叉

核心功能

✅ 状态管理

• 版本化状态模式（JSON模式）
• 完成状态快照
• 切片和部分查询
• 时间步进模拟

✅ 行动系统

10项已实施的行动：

• step -提前时间
• set_resource -设置资源值
• adjust_resource -按增量调整资源
• set_metric -设置度量值
• adjust_metric -按增量调整度量(delta 或 amount 参数）
• initialize -在一次调用中批量初始化资源、指标、标志、实体
• set_flag -设置布尔标志
• add_entity -添加/更新实体
• remove_entity -删除实体
• simulate_load -模拟负载场景（随机）

✅ 世界规则（动态）

• 通过MCP定义JSON规则
• 自动应用 step 行动
• 灵活的条件（比较，和，或不总是）
• 价值来源（资源、指标、标志、元数据、时间、价值）
• 多种操作类型（set_resource、set_metric、set_flag、set_metadata）
• 复杂的公式 使用算术运算：

- 补充： {"type": "add", "values": [...]} - 减法： {"type": "subtract", "left": ..., "right": ...} - 乘法： {"type": "multiply", "values": [...]} - 部门： {"type": "divide", "numerator": ..., "denominator": ...} - 用于复杂计算的嵌套操作

• 规则执行顺序的优先级系统
• 完整规则管理（CRUD操作）

✅ 约束引擎

• 服务器端验证
• 违规时自动回滚状态
• 3+预定义约束：

- NonNegativeResourceConstraint - MaxResourceConstraint - TimeMonotonicConstraint

• 用上下文清除错误消息

✅ 决定论与再现性

• 基于种子的随机数生成
• 相同的种子→ 相同的结果
• 完全可重复的模拟

✅ 审计与可解释性

• 完整的事件历史记录
• 每次变化的状态增量
• 已记录约束检查
• 结构化日志记录（JSON）

✅ 时间轴分叉

• 分支模拟
• 平行的“假设”情景
• 不可变的原始时间线

✅ 坚持

• 保存状态+规则+历史记录
• 管理多个模拟
• 幸存服务器重启
• 设置检查点并恢复
• CRUD操作（保存、加载、列出、删除）
• 不加载元数据检查
• 存储在 ~/.mcp-scenario-engine/simulations/

新增功能

新操作

• adjust_metric -按增量调整度量(delta 或 amount 参数）
• initialize -在一次调用中批量初始化资源、指标、标志、实体

新工具

• get_timeseries -使用变量和时间过滤器查询时间序列快照
• batch_apply_actions -在单个MCP调用中应用多个操作

世界规则增强

• adjust_resource, adjust_metric, adjust_metadata 规则中的动作类型
• clamp_resource, clamp_metric --将值保持在界限内
• min, max, clamp 中的值类型 _compute_value
• priority 和 description 暴露在 add_world_rule

发动机改进

• step 和 steps: N 现在触发世界规则N次（每子步骤一次）
• 用于多步骤调用的每一步时间序列快照
• 时间序列中的手动操作快照（更改状态的非步骤操作）
• reset_simulation 收益 keep_rules 参数
• fork_timeline 收益 activate 参数（使fork成为主动模拟）

持久性修复

• 保存和恢复时间序列快照
• 保存并恢复约束（非负+最大）

安装

先决条件

• Python 3.11+
• pip或uv

本地安装

# Clone repository
git clone https://github.com/schimmmi/mcp-scenario-engine.git
cd mcp-scenario-engine

# Install dependencies
make install

# Or manually
pip install -e ".[dev]"

码头工人

构建和运行演示

# Build Docker image
docker compose build

# Run all demo scenarios
docker compose --profile demo run demo

# Run specific demo
docker compose run demo python examples/demo_scenario_a.py
docker compose run demo python examples/demo_weight_loss.py
docker compose run demo python examples/demo_prisoners_dilemma.py

启动MCP服务器

选项1：独立服务器

# Start server in background
docker compose up -d mcp-scenario-engine

# View logs
docker compose logs -f mcp-scenario-engine

# Stop server
docker compose down

选项2：与Claude Desktop一起使用

添加到您的 claude_desktop_config.json:

{
  "mcpServers": {
    "scenario-engine": {
      "command": "docker",
      "args": [
        "compose",
        "-f",
        "/path/to/mcp-scenario-engine/docker-compose.yml",
        "run",
        "--rm",
        "mcp-scenario-engine"
      ]
    }
  }
}

重要提示：

• 替换 /path/to/mcp-scenario-engine 实际路径
• Docker必须在启动Claude Desktop之前运行
• 模拟保存在容器内（装载一个卷以持久化）

持久挂载量：

# Add to docker-compose.yml under mcp-scenario-engine service:
volumes:
  - ~/.mcp-scenario-engine:/root/.mcp-scenario-engine

这将模拟保存到主目录而不是容器中。

📖 有关完整的Docker文档，请参阅 医生.md

快速开始

1.运行演示

# Both demo scenarios
make demo

# Or individually
python examples/demo_scenario_a.py
python examples/demo_scenario_b.py

2.用作Python库

from mcp_scenario_engine import SimulationEngine
from mcp_scenario_engine.constraints import NonNegativeResourceConstraint

# Create simulation
sim = SimulationEngine(seed=42)

# Set initial state
sim.state.resources = {"budget": 10000.0, "capacity": 100.0}

# Add constraint
sim.constraint_engine.add_constraint(
    NonNegativeResourceConstraint("budget")
)

# Execute actions
result = sim.apply_action(
    "adjust_resource",
    {"resource": "budget", "delta": -2000.0}
)

if result.success:
    print(f"New budget: {sim.state.resources['budget']}")
    print(f"Delta: {result.delta}")
else:
    print(f"Error: {result.message}")
    for v in result.constraints_violated:
        print(f"  - {v.constraint_id}: {v.message}")

3.用作MCP服务器

# Start server
python -m mcp_scenario_engine.server

# Configure in Claude Desktop (claude_desktop_config.json):
{
  "mcpServers": {
    "scenario-engine": {
      "command": "python",
      "args": ["-m", "mcp_scenario_engine.server"],
      "cwd": "/path/to/mcp-scenario-engine"
    }
  }
}

MCP工具

服务器提供20个工具：

状态管理

get_state

获取当前模拟状态。

{}

get_schema

获取状态架构定义。

{}

动作执行

apply_action

执行一个动作。

{
  "action": "adjust_resource",
  "params": {
    "resource": "budget",
    "delta": -500.0
  }
}

模拟控制

reset_simulation

将模拟重置为初始状态。

{
  "seed": 42,
  "keep_rules": true
}

fork_timeline

分叉“假设”场景的时间线。

{
  "name": "optimistic_scenario",
  "activate": true
}

get_history

获取事件历史记录。

{
  "limit": 10
}

世界规则（动态）

add_world_rule

添加自动应用的动态规则 step 行动。

{
  "rule_id": "cpu_overload",
  "condition": {
    "type": "comparison",
    "left": {"type": "resource", "name": "cpu"},
    "operator": ">",
    "right": {"type": "value", "value": 80}
  },
  "actions": [{
    "type": "set_metric",
    "metric": "error_rate",
    "value": {"type": "increment", "amount": 0.05}
  }],
  "priority": 10,
  "description": "Increase error rate on CPU overload"
}

list_world_rules

列出所有活动规则。

{}

get_world_rule

获取特定规则的详细信息。

{
  "rule_id": "cpu_overload"
}

update_world_rule

更新现有规则（部分更新）。

{
  "rule_id": "cpu_overload",
  "priority": 20,
  "description": "Updated description"
}

remove_world_rule

删除规则。

{
  "rule_id": "cpu_overload"
}

clear_world_rules

删除所有规则。

{}

坚持

save_simulation

持久保存模拟（状态+规则+历史）。

{
  "name": "devops_scenario_1",
  "description": "High CPU scenario with 3 steps"
}

load_simulation

加载保存的模拟（替换当前模拟）。

{
  "name": "devops_scenario_1"
}

list_simulations

列出所有已保存的模拟。

{}

get_simulation_info

获取模拟元数据而不加载它。

{
  "name": "devops_scenario_1"
}

delete_simulation

删除已保存的模拟。

{
  "name": "devops_scenario_1"
}

扩展操作

adjust_metric

按增量调整度量。

{"action": "adjust_metric", "params": {"metric": "score", "delta": 5.0}}

initialize

在一次调用中批量初始化多个状态字段。

{
  "action": "initialize",
  "params": {
    "resources": {"budget": 10000, "energy": 100},
    "metrics": {"output": 0, "efficiency": 1.0},
    "flags": {"active": true}
  }
}

批次和时间序列

batch_apply_actions

在单个MCP调用中应用多个操作。

{
  "actions": [
    {"action": "initialize", "params": {"resources": {"budget": 1000}}},
    {"action": "step", "params": {"steps": 5}}
  ],
  "stop_on_failure": true
}

get_timeseries

查询每一步记录的时间序列快照。

{"variables": ["budget", "output"], "from_time": 0, "to_time": 10}

状态模式（v1）

{
  "schema_version": "v1",
  "simulation_id": "uuid",
  "created_at": "2025-01-15T10:30:00Z",
  "updated_at": "2025-01-15T10:35:00Z",
  "seed": 42,
  "time": 0,
  "entities": {},
  "metrics": {},
  "resources": {},
  "flags": {},
  "metadata": {}
}

演示场景

场景A：正常模拟运行

演示：

• 状态初始化
• 多种动作类型
• 资源管理
• 实体生命周期
• 指标跟踪
• 可重复性

python examples/demo_scenario_a.py

输出：

============================================================
DEMO SCENARIO A: Normal Simulation Run
============================================================

1. Creating simulation with seed=42...
2. Adding constraints...
   - Budget must be non-negative
   - Team capacity must be non-negative

3. Executing simulation steps...
   ...
   ✓ Advanced to time step 3

6. Testing Reproducibility:
   ✓ Reproducibility verified - identical results with same seed

场景B：违反约束

演示：

• 约束验证
• 对违规行为进行状态回滚
• 清除错误消息
• 事件历史
• 时间线分叉

python examples/demo_scenario_b.py

输出：

============================================================
DEMO SCENARIO B: Constraint Violation Handling
============================================================

5. Testing constraint violations...

   Violation Attempt 1: Exceed maximum server load
   ✓ REJECTED as expected: Action rejected due to constraint violations
   Violations detected:
     - max_resource_server_load: Resource 'server_load' exceeds maximum 100.0 (got 130.0)
   State unchanged - Server load still: 80.00

演示：世界规则（DevOps）

演示动态规则：

• 基于JSON的规则定义
• 自动因果关系
• 确定性世界模型模拟

python examples/demo_devops_world.py

演示：坚持

演示持久性功能：

• 保存/加载模拟
• 多仿真管理
• 从检查点继续
• 删除和列出操作

python examples/demo_persistence.py

输出：

✅ Persistence Demo Complete!

💡 Key Features:
   • Save simulations with state + rules + history
   • Load simulations and continue from checkpoint
   • List all saved simulations
   • Get metadata without loading
   • Delete simulations
   • Overwrite existing saves

演示：减肥模拟（复杂公式）

演示复杂公式支持：

• 部门： (calorie_deficit / 7700) * 7 * compliance
• 多元乘法
• 加法和减法
• 嵌套操作
• 真实世界的身体成分建模

python examples/demo_weight_loss.py

输出：

Week     Weight   Change      Fat   Muscle    Fat%
----------------------------------------------------------------------
Start    92.47kg        -   18.55kg   70.16kg   20.1%
1        88.53kg   -3.94kg   18.24kg   70.29kg   20.6%
2        88.35kg   -0.18kg   17.93kg   70.41kg   20.3%
...
8        87.26kg   -0.18kg   16.08kg   71.18kg   18.4%

Results: -5.21kg weight, -2.47kg fat, +1.02kg muscle, body fat: 20.1% → 18.4%

演示：博弈论模拟

展示具有复杂战略互动的博弈论模型：

囚徒困境（迭代博弈）

python examples/demo_prisoners_dilemma.py

特征：

• 用收益矩阵迭代囚徒困境
• 以牙还牙vs以牙还牙还牙（相互合作）
• 以牙还牙vs总是有缺陷（报复）
• 计算行动组合收益的战略公式

输出：

Round   P1 Move    P2 Move    P1 Score     P2 Score
----------------------------------------------------------------------
1       Cooperate  Cooperate  3            3
...
10      Cooperate  Cooperate  30           30

Tit-for-Tat vs Always Defect:
1       Cooperate  Defect     0            5
2       Defect     Defect     1            6
...

进化博弈论（鹰鸽）

python examples/demo_evolutionary_game.py

特征：

• 鹰和鸽子的种群动态
• 频率相关的适应度计算
• 复制器动态： hawks_new = hawks * (hawk_fitness / avg_fitness)
• 进化稳定策略（ESS）的出现
• 复杂的嵌套支付公式

输出：

Gen     Hawks   Doves    H_Fit    D_Fit  Avg_Fit
----------------------------------------------------------------------
1        50.0    50.0     12.5     12.5     12.5
...
20       50.0    50.0     12.5     12.5     12.5

Equilibrium: 50% Hawks (matches theoretical ESS!)

拍卖理论（维克雷第二价格）

python examples/demo_auction_theory.py

特征：

• 密封投标二价拍卖
• 讲真话激励示范
• 获胜者的盈余计算： valuation - payment
• 主导战略分析
• 比较真实出价与战略出价

输出：

Winner: Bidder 3
Payment: $95.00 (second-highest bid)
Winner's Surplus: $25.00

Truth-telling test:
• Underbidding → Lost auction, $25 regret
• Overbidding → Same surplus ($25), no benefit

Claude代码工作区

此存储库包括一个具有模拟技能和模板的即用型Claude Code工作区。

快速设置

1. 将此存储库目录用作Claude Code工作区， 或
2. 复制 workspace/ 文件夹和 .claude/ 文件夹到您自己的项目

可用技能（击杀命令）

工作区结构

workspace/
├── CLAUDE.md          # Engine reference & project context
├── templates/         # Simulation templates (health, project, system)
├── designs/           # Pre-simulation design documents
├── results/           # Saved simulation results (examples included)
├── analysis/          # Analysis & insights
├── data/              # Input data & reference values
└── scripts/           # Helper scripts

示例结果

看 workspace/results/ 对于三个工作示例：

• example_spieltheorie.md --Tat vs Always Defect博弈论模拟
• example_sir_epidemie.md --R的SIR流行病模型₀=6
• example_volkswirtschaft.md --三代理凯恩斯经济模拟

测试

# Run all tests
make test

# With coverage
pytest --cov=src --cov-report=html

# Unit tests only
pytest tests/test_*.py -v

# Integration tests only
pytest tests/test_integration.py -v

测试覆盖范围： 80%+（满足要求）

发展

代码质量

# Linting
make lint

# Formatting
make format

# Type checking
mypy src

结构

mcp-scenario-engine/
├── src/mcp_scenario_engine/
│   ├── __init__.py
│   ├── models.py           # Pydantic Models (State, Events)
│   ├── constraints.py      # Constraint Engine
│   ├── actions.py          # Action Implementations
│   ├── simulation.py       # Core Simulation Engine
│   ├── dynamic_rules.py    # Dynamic Rule System
│   ├── world_rules.py      # World Rule Engine
│   ├── persistence.py      # Persistence Layer
│   └── server.py           # MCP Server (20 Tools)
├── tests/
│   ├── test_simulation.py  # Unit Tests
│   ├── test_constraints.py # Constraint Tests
│   ├── test_actions.py     # Action Tests
│   └── test_integration.py # Integration Tests
├── examples/
│   ├── demo_scenario_a.py  # Demo: Normal Simulation
│   ├── demo_scenario_b.py  # Demo: Constraint Violations
│   ├── demo_devops_world.py # Demo: World Rules
│   └── demo_persistence.py  # Demo: Persistence
├── .simulations/            # Saved Simulations
├── Dockerfile
├── docker-compose.yml
├── pyproject.toml
├── Makefile
└── README.md

可观测性

结构化日志记录

所有日志均以JSON格式输出：

{
  "event": "action_applied",
  "simulation_id": "123e4567-e89b-12d3-a456-426614174000",
  "action": "adjust_resource",
  "event_id": "456e7890-e89b-12d3-a456-426614174111",
  "timestamp": "2025-01-15T10:30:00.123456Z"
}

事件类型

• simulation_created -模拟已创建
• simulation_reset -模拟重置
• action_applied -操作成功
• constraint_violated -违反约束
• timeline_forked -时间线分叉

验收标准

✅ 读取状态: get_state 返回有效的架构 ✅ 执行动作: apply_action 返回/delta/event_id之前/之后 ✅ 约束执行：违规会阻止状态更改 ✅ 决定论：相同的种子→ 相同的结果 ✅ 叉/分支：不可变的原始分叉叉

完成的定义

实施✅

• MCP服务器运行（Docker+venv）
• 记录状态模式v1
• 实施了8项行动
• 具有3+规则的约束引擎
• 种子决定论

质量✅

• 单元测试（80%+覆盖率）
• 集成测试（端到端）
• 绒毛/格式化（褶皱/黑色）
• 类型检查（mypy）

可观测性✅

• 结构化日志记录（JSON）
• 日志中没有秘密
• 清除错误消息

文档✅

• 带设置和示例的自述文件
• 工具列表和模式
• 4个演示场景
• 示例输出

演示✅

• make demo 作品
• 场景A：正常运行
• 场景B：约束处理
• 演示：世界规则
• 演示：坚持

许可证

麻省理工学院

联系

如有疑问或问题，请在GitHub上打开问题。