AI模型交互服务器

一个安全、模块化的Python服务器,允许AI模型(如Claude、GPT、Gemini)以受控和安全的方式与计算机交互。

  • by bvirfollet
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 bvirfollet/MCP_server

使用说明

Serveur MCP(模型上下文协议)

纯Python中的安全模块化MCP服务器,允许人工智能模型(Claude、GPT、Gemini)以受控和安全的方式与您的计算机交互。

🎯 概述

MCP服务器公开了AI模型可以调用的“工具”:

  • 读/写文件
  • 执行代码或系统命令
  • 与系统资源交互
  • 访问个性化服务

具有严格的安全性:

  • 客户端身份验证(JWT、mTLS)
  • 每个客户端的粒度权限
  • 沙盒隔离
  • 对所有业务的全面审计

📦 快速启动

先决条件

  • Python 3.10+
  • Git(可选)

安装

# Cloner le repo
git clone https://github.com/yourusername/mcp-server.git
cd mcp-server

# Installer les dépendances de base
pip install -r requirements.txt

# Installer les dépendances de développement
pip install -r requirements-dev.txt

使用简单

基本示例(阶段1)

from mcp_server import MCPServer

# Créer le serveur
server = MCPServer()

# Ajouter un simple tool
@server.tool(
    name="hello",
    description="Salue quelqu'un",
    input_schema={
        "type": "object",
        "properties": {"name": {"type": "string"}},
        "required": ["name"]
    }
)
async def hello_tool(ctx, params):
    name = params.get("name", "World")
    return {"greeting": f"Bonjour {name}!"}

# Démarrer le serveur
if __name__ == "__main__":
    server.run()

具有权限的示例(阶段2)

from mcp_server import MCPServer
from mcp_server.security.permission import Permission, PermissionType

# Créer le serveur
server = MCPServer()

# Outil simple (sans permission)
@server.tool(
    name="greet",
    description="Salue quelqu'un",
    input_schema={"properties": {"name": {"type": "string"}}, "required": ["name"]}
)
async def greet(ctx, params):
    name = params.get("name")
    return {"message": f"Salut {name}!"}

# Outil avec permission FILE_READ
@server.tool(
    name="read_file",
    description="Lit un fichier",
    input_schema={"properties": {"path": {"type": "string"}}, "required": ["path"]},
    permissions=[Permission(PermissionType.FILE_READ, "/app/data/*")]
)
async def read_file(ctx, params):
    path = params.get("path")
    return {"content": f"Contenu de {path}"}

# Démarrer le serveur
if __name__ == "__main__":
    server.run()

另见

  • 演示阶段2 -具有权限的完整MCP客户端
  • 执行 python examples/example_client.py 要查看正在运行的演示:

📚 文档

🏗️ 项目结构

mcp_server/
├── README.md                       # Ce fichier
├── ARCHITECTURE.md                 # Architecture générale
├── IMPLEMENTATION_STRATEGY.md      # Plan de développement
├── SECURITY.md                     # Politique de sécurité
├── CHANGELOG.md                    # Historique des changements
├── pyproject.toml                  # Configuration du projet
├── requirements.txt                # Dépendances
├── requirements-dev.txt            # Dépendances développement
│
├── mcp_server/                     # Code source principal
│   ├── __init__.py
│   ├── core/                       # Cœur du serveur
│   │   ├── mcp_server.py          # Classe serveur principale
│   │   └── constants.py           # Constantes globales
│   ├── transport/                  # Protocoles de transport
│   │   ├── base_transport.py      # Classe abstraite
│   │   ├── stdio_transport.py     # JSON-RPC (stdio)
│   │   ├── tcp_transport.py       # TCP/HTTP+WebSocket
│   │   └── dbus_transport.py      # DBus Linux
│   ├── protocol/                   # Protocole MCP
│   │   ├── mcp_protocol_handler.py
│   │   └── request_router.py
│   ├── security/                   # Sécurité
│   │   ├── permission_manager.py
│   │   ├── authentication_manager.py
│   │   ├── audit_logger.py
│   │   └── client_context.py
│   ├── tools/                      # Gestion des tools
│   │   └── tool_manager.py
│   └── resources/                  # Ressources système
│       ├── filesystem_manager.py
│       ├── execution_manager.py
│       └── sandbox_manager.py
│
├── tests/                          # Tests
│   ├── __init__.py
│   ├── test_integration.py        # Tests d'intégration
│   └── fixtures/                  # Données de test
│
├── docs/                           # Documentation supplémentaire
│   ├── API.md                     # Référence API
│   ├── DEVELOPER_GUIDE.md         # Guide développeurs
│   └── EXAMPLES.md                # Exemples d'utilisation
│
└── .github/
    └── workflows/                  # CI/CD (GitHub Actions)

🚀 发展阶段

阶段 目标 状态
1 服务器启动、基本MCP协议、STDIO传输 ✅ 完整(73次测试)
2 注册和执行工具、RBAC权限、沙盒 ✅ 完整(76次测试)
2.5 代码执行的安全命名空间(可选) ⏳ 即将到来
3 身份验证JWT+持久性JSON(令牌、客户端、审计) ✅ 完成(64+测试)
3.1 可选MTLS和键旋转 ⏳ 即将到来
4 传输TCP/HTTP+WebSocket(多传输并行) ✅ 完成(9项测试)
5 Dbus传输 ⏳ 即将到来
6 过程隔离(子过程) ✅ 完整(51次测试)
7 高级审计和监测 ⏳ 即将到来

📊 验证统计

第一阶段✅

  • 6个模块:传输、协议、客户端上下文、常量、MCPServer
  • 73通过的单元测试
  • 完整的第三方架构

第2阶段✅

  • 6个模块:权限、工具、工具管理器、权限管理器、执行管理器、沙盒上下文
  • 76个通过的单元测试
  • 带审计跟踪的完整RBAC系统
  • 具有持久性变量的每个客户端沙盒上下文
  • 具有超时和验证的安全执行

第三期✅

  • 8个模块:JSONStore、TokenManager、JWTHandler、ClientManager、AuditLogger、MCPServer(第3阶段)、ClientContext(第3部分)、+测试运行器
  • 64+测试单元(JSONStore 8+TokenManager 12+JWTHandler 12+ClientManager 18+AuditLogger 14)
  • 身份验证JWT无状态(HS256)
  • 持久JSON语言环境(tokens.JSON、clients.JSON、audit.JSON)
  • 使用bcrypt密码哈希进行客户管理
  • 审计跟踪不可修改,仅可追加

阶段4✅

  • 2个模块:TCPTransport、WebSocketTransport
  • 9个单元测试(5个TCP+4个WebSocket)
  • 11多传输集成测试
  • TCP支持长度预固定帧
  • 通过AIOHTTP HTTP+升级支持WebSocket

第6阶段✅

  • 4个模块:子进程执行器、客户端隔离管理器、资源管理器、沙盒状态管理器
  • 34个单元测试(7+10+10+7)
  • 17个集成测试(隔离、子流程、配额、状态、权限)
  • 进程隔离执行
  • 每个客户端的隔离目录
  • CPU/内存/磁盘配额
  • 每个客户端的状态持久性
  • 3新生权限RBAC

累积🎯

  • 总计:310+次测试✓ (73 + 76 + 64 + 9 + 11 + 34 + 17)
  • 组件: 实现了32多个模块
  • 架构: 5张沙发(交通→ 协议→ 业务逻辑→ 资源+安全→ 隔离)

🔒 安全

安全是一个方面 批评 这个项目。

主要原则

  • ✅ 默认拒绝 -必须明确授予权限
  • ✅ 审计完成 -记录所有操作
  • ✅ 隔离 -每个客户端都在沙箱中运行
  • ✅ 严格验证 -所有条目均已验证
  • ✅ 加密 -TLS 1.3用于海上运输

类型取消权限

  • FILE_READ -读取文件
  • FILE_WRITE -编写文件
  • FILE_WRITE_GLOBAL -写在范围之外(危险)
  • CODE_EXECUTION -运行Python代码
  • CODE_EXECUTION_SUDO -使用sudo运行
  • SYSTEM_COMMAND -执行系统命令

有关详细信息,请参阅 安全.md.

📝 示例:创建MCP模块

# my_module.py
from mcp_server import Tool, PermissionType, Permission

# Déclarer les tools
class FileReaderTool(Tool):
    """Tool pour lire des fichiers"""

    name = "read_file"
    description = "Lire le contenu d'un fichier"

    input_schema = {
        "type": "object",
        "properties": {
            "path": {"type": "string", "description": "Chemin du fichier"}
        },
        "required": ["path"]
    }

    required_permissions = [
        Permission(PermissionType.FILE_READ, path="*.txt")
    ]

    async def execute(self, client, params):
        path = params.get("path")
        with open(path, 'r') as f:
            return {"content": f.read()}


# Enregistrer le module
def register_module(server):
    server.register_tool(FileReaderTool())


# Dans votre code principal :
# from my_module import register_module
# server = MCPServer()
# register_module(server)

🧪 测试

执行单元测试

# Tous les tests
python -m unittest discover

# Tests spécifiques
python -m unittest tests.test_authentication

执行集成测试

python -m pytest tests/test_integration.py -v

测试用例

coverage run -m unittest discover
coverage report
coverage html  # Rapport HTML dans htmlcov/

📊 高级Niveau建筑

┌─────────────────────────────────────────────────────────┐
│         Client IA (Claude, GPT, Gemini)                 │
└──────────────────────┬──────────────────────────────────┘
                       │
          ┌────────────┴────────────┐
          │                         │
    ┌─────▼──────────────┐   ┌─────▼──────────────┐
    │ Stdio (JSON-RPC)   │   │ TCP/WebSocket      │
    └─────┬──────────────┘   └─────┬──────────────┘
          │                         │
          └────────────┬────────────┘
                       │
        ┌──────────────▼──────────────┐
        │   MCP Protocol Handler      │
        │   Request Router            │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │  Security Layer             │
        │  - Authentication           │
        │  - Authorization (RBAC)     │
        │  - Audit Logging            │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │  Tool Manager               │
        │  - Tool Registry            │
        │  - Parameter Validation     │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │  Execution Manager          │
        │  - Sandboxing               │
        │  - Resource Limiting        │
        │  - Error Handling           │
        └──────────────┬──────────────┘
                       │
        ┌──────────────▼──────────────┐
        │  Resources                  │
        │  - File System              │
        │  - Process Execution        │
        │  - System Calls             │
        └──────────────────────────────┘

🤝 贡献

该项目遵循严格的敏捷方法:

  1. 定义usecase 让我们进行验收测试
  2. 提问者 澄清池的概念
  3. 建筑设计 对于usecase
  4. 实施 le代码和les测试
  5. 验证 验收控制测试
  6. 文档编写者 变化

为了贡献:

  1. Fork le存储库
  2. 为您的UseCase创建分支: git checkout -b feature/uc-xyz
  3. 遵循中定义的敏捷流程 实施_版本.md
  4. 公平取消拉取请求

📋 代码检查表

每次提交前:

  • []所有单元测试都通过
  • []没有未处理的例外
  • []验证验收测试
  • []没有OWASP漏洞
  • []更新的文件
  • []更改日志,填写日期和时间
  • []审核日志已就绪
  • []性能可接受

📞 支持

  • 问题 :打开GitHub问题以查找错误
  • 讨论 :使用讨论提问
  • 安全 :私下报告安全问题

📄 许可证

查看文件 许可证 有关详细信息。

🗓️ 路线图

Q4 2024 (Nov-Déc)
├── Phase 1 : Démarrage serveur
└── Phase 2 : Enregistrement tools

Q1 2025 (Jan-Mar)
├── Phase 3 : Authentification
├── Phase 4 : TCP/HTTP
└── Phase 5 : DBus

Q2 2025 (Avr-Jun)
├── Phase 6 : Sandbox
└── Phase 7 : Audit complet

最后更新: 2025-11-23 版本: 0.1.0-α 状态: 积极发展🚀