UniMCPSim/
├── README.md                 # 项目说明文档
├── requirements.txt          # Python依赖包
├── mcp_server.py            # MCP服务器主程序
├── admin_server.py          # Web管理后台
├── models.py                # 数据库模型定义
├── auth_utils.py            # 认证工具
├── ai_generator.py          # AI响应生成器
├── mcp_client.py            # MCP客户端（Playground用）
├── playground_service.py    # Playground服务
├── logger_utils.py          # 增强日志系统
├── start_servers.py         # 服务启动脚本
├── data/                    # 数据目录
│   └── unimcp.db           # SQLite数据库 (自动创建)
├── logs/                    # 日志目录 (自动创建)
│   ├── unimcp_all.log      # 所有日志
│   ├── unimcp_error.log    # 错误日志
│   └── unimcp_debug.log    # DEBUG详细日志
├── static/                  # 静态资源
│   └── css/
│       └── main.css        # 统一CSS样式
├── templates/               # HTML模板
│   ├── _navigation.html     # 导航组件
│   ├── _footer.html         # 页脚组件
│   ├── login.html           # 登录页
│   ├── dashboard.html       # 仪表板
│   ├── apps.html            # 应用管理
│   ├── tokens.html          # Token管理
│   ├── prompts.html         # 提示词管理
│   ├── playground.html      # MCP Playground
│   ├── logs.html            # 审计日志
│   └── change_password.html # 密码修改
└── tests/                   # 测试文件
    ├── simple_test.py       # 核心功能测试
    └── test_e2e.py         # 端到端测试
├── docs/                    # 文档目录
│   └── images/              # 图片资源
│       ├── diagrams/        # 架构图
│       │   ├── technical-architecture.svg  # 技术架构图 (SVG格式)
│       │   └── technical-architecture.png  # 技术架构图 (PNG格式)
│       └── screenshots/     # 功能截图

# 克隆项目
git clone <repository-url>
cd UniMCPSim

# 创建虚拟环境
python -m venv venv

# 激活虚拟环境
# Linux/Mac:
source venv/bin/activate
# Windows:
venv\Scripts\activate

# 安装依赖
pip install -r requirements.txt

# 设置环境变量（重要！）
unset HTTPS_PROXY
unset HTTP_PROXY

# 创建.env文件
cat > .env << 'EOF'
# ===== AI大模型配置 =====
# 注意：v2.6.0+ 推荐通过Web界面配置，此处配置为备用方式
# 配置优先级：数据库配置（Web界面）> .env文件

# AI API配置（可选 - 推荐通过Web界面配置）
# 支持OpenAI、Qwen、Deepseek、Claude、Gemini等所有兼容OpenAI API的服务
OPENAI_API_KEY=your_api_key_here
OPENAI_MODEL=gpt-4o-mini
OPENAI_API_BASE_URL=https://api.openai.com/v1
OPENAI_ENABLE_THINKING=false  # 禁用思考模式,防止影响JSON输出(默认false)
OPENAI_STREAM=false  # 控制stream模式,某些模型如qwq-32b强制要求true(默认false)

# ===== 服务端口配置 =====
MCP_SERVER_PORT=9090
ADMIN_SERVER_PORT=9091

# ===== 日志配置 =====
DEBUG=false              # 启用DEBUG模式记录详细调用信息
LOG_LEVEL=INFO          # 日志级别: DEBUG, INFO, WARNING, ERROR
LOG_DIR=logs            # 日志文件存储目录
EOF

python start_servers.py

# 启动MCP服务器（端口9090）
python mcp_server.py

# 启动Web管理后台（端口9091）
python admin_server.py

==========================================================
UniMCPSim - Universal MCP Simulator
==========================================================

初始化默认模拟器...
✅ 已创建默认管理员用户

启动服务...
服务已启动:
----------------------------------------------------------
MCP服务器: http://localhost:9090/<Category>/<Product>?token=<token>
管理后台: http://localhost:9091/admin/
默认账号: admin / admin123
----------------------------------------------------------

按 Ctrl+C 停止服务

# 构建并启动
docker compose up -d --build

# 查看状态
docker compose ps

# 查看日志
docker compose logs -f unimcpsim

./data  -> /app/data   # SQLite数据库
./logs  -> /app/logs   # 运行日志

docker pull ghcr.io/flagify-com/unimcpsim:latest
docker run -d --name unimcpsim \
  --restart unless-stopped \
  -p 9090:9090 \
  -p 9091:9091 \
  -v $(pwd)/data:/app/data \
  -v $(pwd)/logs:/app/logs \
  -e TZ=Asia/Shanghai \
  ghcr.io/flagify-com/unimcpsim:latest

docker pull ghcr.io/flagify-com/unimcpsim:sha-8ff8e19

docker pull ghcr.io/flagify-com/unimcpsim:sha-8ff8e19

docker save ghcr.io/flagify-com/unimcpsim:sha-8ff8e19 \
  | gzip > unimcpsim-sha-8ff8e19.tar.gz

sha256sum unimcpsim-sha-8ff8e19.tar.gz > SHA256SUMS

sha256sum -c SHA256SUMS
gunzip -c unimcpsim-sha-8ff8e19.tar.gz | docker load

docker tag ghcr.io/flagify-com/unimcpsim:sha-8ff8e19 unimcpsim:sha-8ff8e19

docker run -d --name unimcpsim \
  --restart unless-stopped \
  -p 9090:9090 \
  -p 9091:9091 \
  -v /opt/unimcpsim/data:/app/data \
  -v /opt/unimcpsim/logs:/app/logs \
  -e TZ=Asia/Shanghai \
  unimcpsim:sha-8ff8e19

docker tag ghcr.io/flagify-com/unimcpsim:sha-8ff8e19 \
  harbor.example.com/security/unimcpsim:sha-8ff8e19

docker push harbor.example.com/security/unimcpsim:sha-8ff8e19

docker build \
  --build-arg USE_CHINA_MIRRORS=true \
  --build-arg PYTHON_IMAGE=python:3.11-slim \
  -t unimcpsim:local .

# 如果内网已有基础镜像，也可以复用：
docker build \
  --build-arg USE_CHINA_MIRRORS=true \
  --build-arg PYTHON_IMAGE=deepsocx-backend:latest \
  -t unimcpsim:local .

cd UniMCPSim
source venv/bin/activate
unset HTTPS_PROXY && unset HTTP_PROXY

# 运行核心功能测试
python tests/simple_test.py

############################################################
# UniMCPSim 功能验证测试
############################################################

============================================================
测试应用列表功能
============================================================
✅ 找到 9 个已启用的应用:
   - VirusTotal威胁情报 (/Security/VirusTotal)
   - 微步在线威胁情报 (/Security/ThreatBook)
   - 青藤云HIDS (/Security/QingTengHIDS)
   - 企业微信 (/IM/WeChat)
   - 腾讯会议 (/Meeting/TencentMeeting)
   - Jira工单系统 (/Ticket/Jira)
   - 华为交换机 (/Network/HuaweiSwitch)
   - Cisco路由器 (/Network/CiscoRouter)
   - 深信服防火墙 (/Firewall/Sangfor)

============================================================
UniMCPSim 核心功能测试
============================================================
✅ 使用Token: dbb02af0...a525

测试 1: 企业微信发送消息 ✅ 成功
测试 2: VirusTotal IP扫描 ✅ 成功
测试 3: Jira创建工单 ✅ 成功
测试 4: 深信服防火墙封禁IP ✅ 成功
测试 5: 华为交换机查看接口 ✅ 成功

============================================================
测试总结
============================================================
🎉 所有测试通过!

# 获取Demo Token - 方法1：从管理后台复制
# 访问 http://localhost:9091/admin/tokens 复制Demo Token

# 获取Demo Token - 方法2：使用脚本
TOKEN=$(python -c "
import sys
sys.path.append('.')
from models import db_manager, Token
session = db_manager.get_session()
try:
    demo_token = session.query(Token).filter_by(name='Demo Token').first()
    if demo_token:
        print(demo_token.token)
finally:
    session.close()
")

# 测试企业微信发送消息
curl "http://localhost:9090/IM/WeChat?token=$TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "send_text_message",
    "parameters": {
      "to_user": "test_user",
      "content": "Hello from UniMCPSim!"
    }
  }'

# 获取Demo Token (在管理后台获取或使用脚本)
TOKEN="demo-token-here"

# 直接调用企业微信API端点
curl "http://localhost:9090/IM/WeChat?token=$TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "send_text_message",
    "parameters": {
      "to_user": "test_user",
      "content": "Hello from UniMCPSim!"
    }
  }'

curl "http://localhost:9090/Security/VirusTotal?token=$TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "scan_ip",
    "parameters": {
      "ip": "8.8.8.8"
    }
  }'

curl "http://localhost:9090/Ticket/Jira?token=$TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "create_issue",
    "parameters": {
      "project_key": "TEST",
      "issue_type": "Task",
      "summary": "测试工单",
      "description": "这是一个测试工单",
      "priority": "High"
    }
  }'

curl "http://localhost:9090/Firewall/Sangfor?token=$TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "action": "block_ip_address",
    "parameters": {
      "ip_address": "192.168.1.100",
      "reason": "可疑活动"
    }
  }'

{
  "mcpServers": {
    "QAXFW": {
      "name": "奇安信防火墙",
      "type": "streamableHttp",
      "description": "奇安信防火墙MCP服务",
      "isActive": true,
      "baseUrl": "http://127.0.0.1:9090/Firewall/QAXFW?token=YOUR_TOKEN_HERE"
    }
  }
}

[
  {
    "name": "your_action",
    "display_name": "动作显示名称",
    "description": "动作描述",
    "parameters": [
      {
        "key": "param_name",
        "type": "String",
        "required": true,
        "description": "参数描述"
      }
    ]
  }
]

http://localhost:9090/{Category}/{ProductName}?token=<your-token>

curl -X POST "http://localhost:9090/IM/WeChat?token=f1bb3770-6e46-4fe6-b518-e1c738c7b6a4" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "method": "initialize",
    "params": {
      "protocolVersion": "0.1.0",
      "capabilities": {},
      "clientInfo": {
        "name": "demo-client",
        "version": "1.0.0"
      }
    },
    "id": 1
  }' \
  -D headers.txt

event: message
data: {
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "protocolVersion": "2025-06-18",
    "capabilities": {
      "experimental": {},
      "prompts": {"listChanged": true},
      "resources": {"subscribe": false, "listChanged": true},
      "tools": {"listChanged": true}
    },
    "serverInfo": {
      "name": "UniMCPSim",
      "version": "1.0.0"
    },
    "instructions": "Universal MCP Simulator - 通用MCP模拟器，可动态模拟各种产品的API接口"
  }
}

session_id=$(grep -i 'mcp-session-id:' headers.txt | tr -d '\r' | cut -d' ' -f2-)
echo "会话ID: $session_id"

# 使用提取的会话ID
curl -X POST "http://localhost:9090/IM/WeChat?token=f1bb3770-6e46-4fe6-b518-e1c738c7b6a4" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "mcp-session-id: $session_id" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tools/call",
    "params": {
      "name": "execute_action",
      "arguments": {
        "token": "f1bb3770-6e46-4fe6-b518-e1c738c7b6a4",
        "category": "IM",
        "product": "WeChat",
        "action": "send_message",
        "parameters": {
          "to_user": "zhang.san",
          "text": "项目会议将于明天下午2点在A会议室举行，请准时参加。"
        }
      }
    },
    "id": 2
  }'

import asyncio
import httpx
import json

class UniMCPSimClient:
    def __init__(self, base_url="http://localhost:9090", token=None, product_path=None):
        self.base_url = base_url
        self.token = token
        self.product_path = product_path  # 例如: "IM/WeChat"
        self.session_id = None

    @property
    def endpoint_url(self):
        """获取完整的端点URL"""
        if self.product_path:
            return f"{self.base_url}/{self.product_path}"
        else:
            return f"{self.base_url}/mcp"

    async def initialize(self):
        """初始化MCP连接"""
        async with httpx.AsyncClient() as client:
            payload = {
                "jsonrpc": "2.0",
                "method": "initialize",
                "params": {
                    "protocolVersion": "0.1.0",
                    "capabilities": {},
                    "clientInfo": {"name": "python-client", "version": "1.0.0"}
                },
                "id": 1
            }

            response = await client.post(
                f"{self.endpoint_url}?token={self.token}",
                json=payload,
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream"
                }
            )

            # 提取会话ID
            self.session_id = response.headers.get('mcp-session-id')
            return response.status_code == 200

    async def execute_action(self, category, product, action, parameters):
        """执行模拟器动作"""
        if not self.session_id:
            raise Exception("请先调用initialize()初始化连接")

        async with httpx.AsyncClient() as client:
            payload = {
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {
                    "name": "execute_action",
                    "arguments": {
                        "token": self.token,
                        "category": category,
                        "product": product,
                        "action": action,
                        "parameters": parameters
                    }
                },
                "id": 2
            }

            response = await client.post(
                f"{self.endpoint_url}?token={self.token}",
                json=payload,
                headers={
                    "Content-Type": "application/json",
                    "Accept": "application/json, text/event-stream",
                    "mcp-session-id": self.session_id
                }
            )

            return response.text

# 使用示例
async def main():
    # 使用产品特定端点（推荐）
    client = UniMCPSimClient(
        token="f1bb3770-6e46-4fe6-b518-e1c738c7b6a4",
        product_path="IM/WeChat"
    )

    # 初始化连接
    if await client.initialize():
        print("✅ MCP连接初始化成功")

        # 企业微信发送消息
        result = await client.execute_action(
            category="IM",
            product="WeChat",
            action="send_message",
            parameters={
                "to_user": "zhang.san",
                "text": "Python客户端测试消息"
            }
        )
        print("发送消息结果:", result)
    else:
        print("❌ MCP连接初始化失败")

# 运行示例
if __name__ == "__main__":
    asyncio.run(main())

# 运行核心功能测试
python tests/simple_test.py

# 运行端到端测试（需要服务器运行）
python tests/test_e2e.py

#!/bin/bash

# 设置变量
MCP_SERVER="http://localhost:9090/IM/WeChat"
TOKEN="f1bb3770-6e46-4fe6-b518-e1c738c7b6a4"

echo "=== UniMCPSim 企业微信测试 ==="

# 步骤1：初始化连接
echo "1. 初始化MCP连接..."
init_response=$(curl -s -X POST "${MCP_SERVER}?token=${TOKEN}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -d '{
    "jsonrpc": "2.0",
    "method": "initialize",
    "params": {
      "protocolVersion": "0.1.0",
      "capabilities": {},
      "clientInfo": {"name": "test-client", "version": "1.0.0"}
    },
    "id": 1
  }' \
  -D /tmp/headers.txt)

# 提取会话ID
session_id=$(grep -i 'mcp-session-id:' /tmp/headers.txt | tr -d '\r' | cut -d' ' -f2-)

if [ -z "$session_id" ]; then
    echo "❌ 获取会话ID失败"
    exit 1
fi

echo "✅ 连接成功，会话ID: $session_id"

# 步骤2：发送消息
echo ""
echo "2. 测试企业微信发送消息..."
msg_response=$(curl -s -X POST "${MCP_SERVER}?token=${TOKEN}" \
  -H "Content-Type: application/json" \
  -H "Accept: application/json, text/event-stream" \
  -H "mcp-session-id: $session_id" \
  -d '{
    "jsonrpc": "2.0",
    "method": "tools/call",
    "params": {
      "name": "execute_action",
      "arguments": {
        "token": "'$TOKEN'",
        "category": "IM",
        "product": "WeChat",
        "action": "send_message",
        "parameters": {
          "to_user": "test.user",
          "text": "这是一条测试消息"
        }
      }
    },
    "id": 2
  }')

if echo "$msg_response" | grep -q "event: message"; then
    echo "✅ 消息发送成功"
    echo "$msg_response" | grep "data:" | sed 's/data: //'
else
    echo "❌ 消息发送失败"
    echo "$msg_response"
fi

echo ""
echo "=== 测试完成 ==="

chmod +x test_wechat.sh
./test_wechat.sh

# 检查端口占用
lsof -i :9090
lsof -i :9091

# 杀死占用进程
kill -9 <PID>

# 升级pip
pip install --upgrade pip

# 清除缓存重新安装
pip cache purge
pip install -r requirements.txt

# 完全重置
rm -rf data/
mkdir data
python init_simulators.py

# 确保取消代理设置
unset HTTPS_PROXY
unset HTTP_PROXY
unset http_proxy
unset https_proxy

# 验证
echo $HTTPS_PROXY  # 应该为空

python mcp_server.py

# 通过Web管理后台
curl "http://localhost:9091/admin/api/tokens" -H "Authorization: Bearer <session>"

# 或直接查询数据库
python -c "
from models import db_manager, Token
session = db_manager.get_session()
tokens = session.query(Token).filter_by(enabled=True).all()
for token in tokens:
    print(f'{token.name}: {token.token}')
session.close()
"

# 停止服务 (Ctrl+C)
# 重新启动
python start_servers.py

# 删除现有数据库
rm -f data/unimcp.db

# 重新初始化
python init_simulators.py

@mcp.tool()
async def your_new_tool(param1: str, param2: int) -> str:
    """工具描述"""
    # 实现逻辑
    return "result"

2025-09-30 16:05:12 - INFO - MCP Request: POST:tools/list /ThreatIntelligence/Threatbook - SUCCESS

2025-09-30 16:10:30 - INFO - AI Call: OpenAI/qwen3-max - SUCCESS (duration: 1.23s)

2025-09-30 16:12:15 - INFO - Tool Call: query_ip_threat_intel (app: ThreatIntelligence/Threatbook) - SUCCESS

2025-09-30 16:15:00 - WARNING - Auth Failure: Token required (path: /IM/WeChat, ip: 127.0.0.1)

# 在 .env 文件中设置
DEBUG=true
LOG_LEVEL=DEBUG