MCP模块化架构

模型上下文协议（MCP）的生产就绪参考实现，具有干净的分层架构，专为可扩展性和可维护性而设计。

](https://www.python.org/downloads/) ![License](LICENSE) ![Tests](tests/) ![Coverage](docs/TESTING.md)

______________________________________________________________________

概述

MCP模块化架构 是一个参考实现，展示了构建基于MCP的系统的最佳实践，包括：

干净的分层架构 严格分离关注点
运输抽象 实现与协议无关的服务器实现
SDK首次设计 便于客户端集成
综合测试 核心业务逻辑的代码覆盖率>95%
零硬编码配置 使用基于YAML的配置管理

什么是MCP？

模型上下文协议（MCP）是AI代理与外部工具、资源和提示模板交互的协议。该项目实现了一个完整的MCP服务器，包括所有三种基本类型：

工具 --可执行函数（例如计算器、文件操作）
资源 --可读取的数据源（例如，配置、系统状态）
提示 --LLM交互的预定义提示模板

这是给谁的？

开发者 构建MCP服务器或客户端
建筑师 寻求清洁架构模式的参考实现
团队为AI代理系统寻找模块化、可测试的基础

______________________________________________________________________

建筑

该系统遵循严格的 层次结构 具有单向依赖性：

┌─────────────────────────────────────────┐
│          User Interface (CLI)            │
│     ↓ uses                              │
├─────────────────────────────────────────┤
│            Client SDK                    │
│     ↓ uses                              │
├─────────────────────────────────────────┤
│         Transport Layer                  │
│     (STDIO, HTTP, WebSocket)            │
│     ↓ uses                              │
├─────────────────────────────────────────┤
│           MCP Server                     │
│  (Tools, Resources, Prompts)            │
│     ↓ uses                              │
├─────────────────────────────────────────┤
│       Core Infrastructure                │
│  (Config, Logging, Errors)              │
└─────────────────────────────────────────┘

层职责

1.核心基础设施

整个应用程序中使用的基础服务：

配置管理 --基于YAML的环境感知配置
日志记录 --具有文件旋转和控制台输出的结构化日志记录
错误处理 --自定义异常层次结构和集中错误处理

2.MCP服务器层

实现模型上下文协议的业务逻辑：

服务器 --管理生命周期并协调图元
工具注册表 --集中工具注册和发现
资源注册表 --基于URI的资源管理
快速注册 --基于模板的提示管理

3.传输层

与协议无关的通信：

基地运输 --所有传输实现的抽象接口
STDIO传输 --标准输入/输出通信（建议用于MCP）
运输搬运员 --在传输和MCP服务器之间路由消息

4.SDK层

MCP服务器集成的客户端库：

MCP客户端 -高级API包装运输通信
上下文管理器支持连接生命周期
具有错误检测功能的自动请求/响应处理

5.用户界面

终端用户交互：

命令行界面 --使用SDK的命令行界面
所有MCP操作的用户友好命令
JSON参数支持和格式化输出

______________________________________________________________________

关键概念

工具

工具是执行动作的可执行函数。每个工具：

为输入参数定义JSON模式
实现一个 execute() 方法
返回标准化的结果格式

示例工具：

calculator --执行算术运算（加、减、乘、除）
echo --简单的回声功能

资源

资源是由URI标识的数据源。资源可以是：

静态 --内容保持不变（例如，配置文件）
动态的 --每次读取时内容都会发生变化（例如，系统状态）

示例资源：

config://app --应用程序配置
status://system --带时间戳的系统状态

提示

提示是LLM交互的模板。每个提示：

接受参数（必需和可选）
返回消息数组（系统、用户、助手）
支持基于模板的消息生成

示例提示：

code_review --指导模型审查质量代码
summarize --引导模型总结文本

运输抽象

传输层是 完全解耦 来自MCP逻辑：

MCP服务器对传输机制一无所知
在不更改MCP代码的情况下，将STDIO替换为HTTP/WebSocket
运输处理器将两层连接起来

SDK优先设计

客户端SDK提供 洁净、高水平的API:

UI组件仅使用SDK（从不直接传输或MCP）
SDK适用于任何传输实现
无业务逻辑重复

______________________________________________________________________

项目结构

mcp-modular-architecture/
├── config/                      # Configuration files
│   ├── base.yaml                # Base configuration
│   ├── development.yaml         # Development environment
│   └── production.yaml          # Production environment
│
├── src/                         # Source code
│   ├── core/                    # Core infrastructure
│   │   ├── config/              # Configuration management
│   │   ├── logging/             # Logging system
│   │   └── errors/              # Error handling
│   │
│   ├── mcp/                     # MCP server layer
│   │   ├── server.py            # MCP server
│   │   ├── tool_registry.py     # Tool registry
│   │   ├── resource_registry.py # Resource registry
│   │   ├── prompt_registry.py   # Prompt registry
│   │   ├── tools/               # Tool implementations
│   │   ├── resources/           # Resource implementations
│   │   ├── prompts/             # Prompt implementations
│   │   └── schemas/             # JSON schemas
│   │
│   ├── transport/               # Transport layer
│   │   ├── base_transport.py    # Abstract transport
│   │   ├── stdio_transport.py   # STDIO transport
│   │   └── transport_handler.py # Message routing
│   │
│   ├── sdk/                     # Client SDK
│   │   └── mcp_client.py        # MCP client
│   │
│   ├── ui/                      # User interface
│   │   └── cli.py               # CLI interface
│   │
│   ├── models/                  # Domain models
│   ├── services/                # Service layer
│   └── utils/                   # Utilities
│
├── tests/                       # Unit tests (165 tests)
│   ├── core/
│   ├── mcp/
│   ├── transport/
│   ├── sdk/
│   └── ...
│
├── docs/                        # Documentation
├── pyproject.toml               # Project metadata
└── requirements.txt             # Dependencies

______________________________________________________________________

安装

先决条件

Python 3.10 或更高
点（Python包管理器）

设置

克隆存储库：

   git clone https://github.com/TalBarda8/mcp-modular-architecture.git
   cd mcp-modular-architecture

创建虚拟环境 （推荐）：

   python -m venv venv
   source venv/bin/activate  # On Windows: venv\Scripts\activate

安装依赖项：

   pip install -r requirements.txt

或者使用开发依赖项进行安装：

   pip install -e ".[dev]"

______________________________________________________________________

快速开始

3分钟后起床跑步：

# 1. Clone and install
git clone https://github.com/TalBarda8/mcp-modular-architecture.git
cd mcp-modular-architecture
pip install -r requirements.txt

# 2. Run tests
pytest

# 3. Try the programmatic API
python3 -c "
from src.mcp.server import MCPServer
from src.mcp.tools.calculator_tool import CalculatorTool

server = MCPServer()
server.initialize(tools=[CalculatorTool()])
result = server.execute_tool('calculator', {'operation': 'add', 'a': 5, 'b': 3})
print(f\"Result: {result['result']['result']}\")
"

______________________________________________________________________

使用模式

此库支持三种使用模式：

1.嵌入式服务器（库使用）

将MCP服务器直接嵌入到您的应用程序中：

from src.mcp.server import MCPServer
from src.mcp.tools.calculator_tool import CalculatorTool

server = MCPServer()
server.initialize(tools=[CalculatorTool()])
result = server.execute_tool('calculator', {'operation': 'add', 'a': 10, 'b': 5})

在以下情况下使用此功能：构建包含MCP功能的自定义应用程序。

2.独立服务器+CLI

将服务器作为独立进程运行，并通过CLI进行交互：

# Terminal 1: Start the server
python run_server.py

# Terminal 2: Use the CLI
python -m src.ui.cli info
python -m src.ui.cli tool calculator --params '{"operation": "add", "a": 10, "b": 5}'

在以下情况下使用此功能：测试CLI或构建客户端应用程序。

3.独立服务器+SDK

在某个进程中运行服务器，在另一个进程中连接SDK：

from src.sdk.mcp_client import MCPClient
from src.transport.stdio_transport import STDIOTransport
import subprocess

# Start server as subprocess
server_process = subprocess.Popen(
    ['python', 'run_server.py'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    text=True
)

# Connect with SDK
transport = STDIOTransport()
transport._input_stream = server_process.stdout
transport._output_stream = server_process.stdin

client = MCPClient(transport)
# Use client...

在以下情况下使用此功能：构建连接到外部MCP服务器的程序化客户端。

______________________________________________________________________

运行项目

配置

使用设置环境 APP_ENV 环境变量：

# Development (default)
export APP_ENV=development

# Production
export APP_ENV=production

创建一个 config/local.yaml 本地覆盖文件（gitignored）：

logging:
  level: "DEBUG"

运行独立服务器

这 run_server.py 脚本启动侦听STDIO的MCP服务器：

python run_server.py

服务器将：

使用所有内置工具、资源和提示进行初始化
在stdin上监听JSON-RPC消息
将响应发送到stdout
运行直到中断（Ctrl+C）

这有助于：

测试CLI
开发客户端应用程序
集成测试

以编程方式使用MCP服务器（嵌入式）

from src.mcp.server import MCPServer
from src.mcp.tools.calculator_tool import CalculatorTool
from src.mcp.tools.echo_tool import EchoTool
from src.mcp.resources.config_resource import ConfigResource
from src.mcp.resources.status_resource import StatusResource
from src.mcp.prompts.code_review_prompt import CodeReviewPrompt
from src.mcp.prompts.summarize_prompt import SummarizePrompt

# Initialize server
server = MCPServer()

# Register primitives
server.initialize(
    tools=[CalculatorTool(), EchoTool()],
    resources=[ConfigResource(), StatusResource()],
    prompts=[CodeReviewPrompt(), SummarizePrompt()]
)

# Execute a tool
result = server.execute_tool('calculator', {
    'operation': 'add',
    'a': 10,
    'b': 5
})
print(result)  # {'success': True, 'result': {'result': 15}}

# Read a resource
config = server.read_resource('config://app')
print(config)

# Get prompt messages
messages = server.get_prompt_messages('code_review', {
    'code': 'def foo(): pass',
    'language': 'python'
})
print(messages)

______________________________________________________________________

CLI使用情况

CLI提供了一个用户友好的界面，用于与正在运行的MCP服务器进行交互。

先决条件：在单独的终端中启动MCP服务器：

python run_server.py

可用命令

# Show server information
python -m src.ui.cli info

# List all tools
python -m src.ui.cli tools

# Execute a tool
python -m src.ui.cli tool calculator --params '{"operation": "add", "a": 10, "b": 5}'

# List all resources
python -m src.ui.cli resources

# Read a resource
python -m src.ui.cli resource config://app

# List all prompts
python -m src.ui.cli prompts

# Get prompt messages
python -m src.ui.cli prompt code_review --args '{"code": "def foo(): pass", "language": "python"}'

完整示例

# Terminal 1: Start the server
$ python run_server.py
2025-12-26 11:00:00 - ServerRunner - INFO - MCP server ready. Listening on STDIO...

# Terminal 2: Use the CLI
$ python -m src.ui.cli info
Server: MCP Modular Architecture Server v2.0.0
Status: Running
Capabilities: tools, resources, prompts

$ python -m src.ui.cli tools
Available tools:
  - calculator: Perform basic arithmetic operations
  - echo: Echo input message

$ python -m src.ui.cli tool calculator --params '{"operation": "multiply", "a": 7, "b": 6}'
Success: true
Result: 42

备注：CLI通过STDIO传输连接到服务器。每个CLI命令都会向服务器发送JSON-RPC请求并显示响应。

______________________________________________________________________

SDK使用

SDK提供了一个干净、高级的API，用于与MCP服务器集成。

连接到外部服务器

SDK连接到正在运行的MCP服务器进程：

from src.sdk.mcp_client import MCPClient
from src.transport.stdio_transport import STDIOTransport
import subprocess

# Start server as a subprocess
server_process = subprocess.Popen(
    ['python', 'run_server.py'],
    stdin=subprocess.PIPE,
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True
)

# Create transport that communicates with the server process
transport = STDIOTransport()
transport._input_stream = server_process.stdout
transport._output_stream = server_process.stdin

# Create client
client = MCPClient(transport)

try:
    # Connect to server
    client.connect()

    # Get server info
    info = client.get_server_info()
    print(f"Connected to {info['name']} v{info['version']}")

    # List available tools
    tools = client.list_tools()
    print(f"Available tools: {tools}")

    # Execute a tool
    result = client.execute_tool('calculator', {
        'operation': 'add',
        'a': 5,
        'b': 3
    })
    print(f"Result: {result['result']['result']}")

    # Read a resource
    config = client.read_resource('config://app')
    print(f"Config loaded: {len(config['content'])} keys")

finally:
    # Clean up
    client.disconnect()
    server_process.terminate()
    server_process.wait()

与现有服务器一起使用

如果您的服务器已在另一个终端中运行：

from src.sdk.mcp_client import MCPClient
from src.transport.stdio_transport import STDIOTransport
import sys

# Note: This requires the server to be running in the same process context
# For production use, use the subprocess approach above
transport = STDIOTransport()
client = MCPClient(transport)

with client:
    tools = client.list_tools()
    print(f"Available tools: {[t['name'] for t in tools]}")

SDK方法

服务器方法：

get_server_info() --获取服务器信息
initialize_server() --初始化服务器

工具方法：

list_tools() --列出可用工具
execute_tool(name, parameters) --执行工具

资源方法：

list_resources() --列出可用资源
read_resource(uri) --按URI读取资源

提示方法：

list_prompts() --列出可用提示
get_prompt_messages(name, arguments) --获取提示消息

______________________________________________________________________

运行测试

该项目包括全面的单元测试，包括 >95%的代码覆盖率 核心业务逻辑。

运行所有测试

pytest

跑步有保障

pytest --cov=src --cov-report=html --cov-report=term-missing

查看覆盖率报告： open htmlcov/index.html

运行特定测试

# Run tests for a specific module
pytest tests/mcp/test_server.py

# Run tests for SDK
pytest tests/sdk/

# Run tests with verbose output
pytest -v

# Run tests matching a pattern
pytest -k "test_calculator"

检验统计量

总测试： 190
通过率： 100%
新闻报道： 95.12%（核心业务逻辑，不包括UI层）
测试机构： 测试镜像源结构

保险范围详细信息

该项目对所有核心业务逻辑保持>95%的单元测试覆盖率：

✅ 核心基础设施 （配置、日志记录、错误）：93%+
✅ MCP服务器 （工具、资源、提示）：95%+
✅ 传输层: 85%+
✅ 软件开发工具包 （MCP客户端）：100%
✅ 模型和实用程序: 100%

注： UI层(src/ui/)有意排除在单元测试覆盖范围之外。CLI/UI代码最好通过集成测试、E2E测试或手动测试进行测试。看 docs/TEST.md 详细的测试策略和原理。

______________________________________________________________________

视觉示例

有关全面的可视化文档和实际使用示例，请参阅 docs/screenshots.md.

快速概览

该项目包括真实的工作示例，展示了：

1. SDK演示 -完成工作流程

显示服务器初始化、工具列表和两个并行处理工具的执行：

1. Initializing MCP Server...
   ✓ Server initialized

2. Listing Available Tools...
   • calculator, echo, batch_processor, concurrent_fetcher

3. Executing batch_processor (Multiprocessing)...
   Results: 5 items processed, Workers used: 2

4. Executing concurrent_fetcher (Multithreading)...
   Results: 3 items processed, Threads used: 3

运行它: export PYTHONPATH=. && python3 examples/sdk_demo.py

2. 多处理（CPU限制）

18个测试演示 multiprocessing.Pool 对于并行处理：

CPU内核之间的真正并行性
绕过Python的GIL
0.80s内测试通过率100%

3. 多线程（I/O绑定）

20项测试证明 ThreadPoolExecutor 对于并发I/O：

I/O等待期间并发执行
无锁螺纹安全
包括加速验证测试

4. 完整测试套件

228测试 3.09秒通过
95.29%的代码覆盖率
包括两种并行处理方法

查看完整的屏幕截图和输出: docs/screenshots.md

______________________________________________________________________

插件示例：系统可扩展性

一个完整的工作插件示例演示了系统的可扩展性，而无需修改核心代码。

WeatherTool插件

位置: examples/plugins/weather_plugin.py

此插件显示了如何使用外部工具扩展MCP：

无需更改核心代码 （演示开放/封闭原理）
使用现有扩展点 （BaseTool、ToolRegistry）
工作原理与内置工具完全相同 （相同的初始化、执行、列表）

运行插件演示

# Run plugin demo
export PYTHONPATH=.
python3 examples/plugins/plugin_demo.py

输出:

MCP Plugin Demo - System Extensibility

1. Initializing MCP Server...
   ✓ Server initialized with built-in tools + weather plugin

2. Listing All Available Tools (Built-in + Plugin)...
   • [Built-in] calculator: Perform basic arithmetic operations...
   • [Built-in] echo: Echo back the provided message...
   • [Built-in] batch_processor: Process a batch of numbers in parallel...
   • [Built-in] concurrent_fetcher: Process items concurrently...
   • [PLUGIN  ] weather: Get current weather information for a city...

4. Testing Plugin Tool (weather)...
   City: Tel Aviv
   Temperature: 22°C
   Condition: Rainy
   Humidity: 61%

关键要点

✓ 零核心修改：插件完全是外部的 ✓ 相同的界面：插件的工作方式类似于内置工具 ✓ 干净的建筑：使用依赖关系反转（BaseTool抽象） ✓ 开闭原则：系统开放扩展，关闭修改

创建自己的插件

📖 插件开发指南 -全面的分步指南

本指南包括：

分步插件创建教程
所需接口和扩展点
常见错误以及如何避免
测试、命名和注册的最佳实践
使用代码片段完成工作示例

其他文件:

建筑细节：第9.4.3节 docs/architecture.md
示例代码：示例/插件/

______________________________________________________________________

发展

插件开发

有关全面的插件开发指导，请参阅：

📖 插件开发指南

这包括分步说明、常见陷阱、最佳实践和完整示例。

添加新工具

创建工具类 继承自 BaseTool:

from src.mcp.tools.base_tool import BaseTool

class MyTool(BaseTool):
    def __init__(self):
        super().__init__(
            name="my_tool",
            description="Description of my tool",
            input_schema={
                "type": "object",
                "properties": {
                    "param1": {"type": "string"}
                },
                "required": ["param1"]
            }
        )

    def execute(self, parameters: dict) -> dict:
        # Implementation
        return {"result": "some value"}

注册工具 与服务器：

from src.mcp.server import MCPServer
from my_tool import MyTool

server = MCPServer()
server.initialize(tools=[MyTool()])

添加新传输

创建传输类 继承自 BaseTransport:

from src.transport.base_transport import BaseTransport

class HTTPTransport(BaseTransport):
    def start(self) -> None:
        # Start HTTP server
        pass

    def stop(self) -> None:
        # Stop HTTP server
        pass

    def send_message(self, message: dict) -> None:
        # Send HTTP response
        pass

    def receive_message(self) -> dict:
        # Receive HTTP request
        pass

使用交通工具 使用SDK：

from src.sdk.mcp_client import MCPClient
from my_transport import HTTPTransport

transport = HTTPTransport()
client = MCPClient(transport)

添加新资源

创建资源类 继承自 BaseResource:

from src.mcp.resources.base_resource import BaseResource

class MyResource(BaseResource):
    def __init__(self):
        super().__init__(
            uri="custom://my-resource",
            name="My Resource",
            description="Description of resource",
            mime_type="application/json"
        )

    def read(self) -> dict:
        return {
            "uri": self.uri,
            "content": {"key": "value"}
        }

    def is_dynamic(self) -> bool:
        return False  # True if content changes

注册资源 与服务器：

server.initialize(resources=[MyResource()])

扩展CLI

编辑 src/ui/cli.py 添加新命令。CLI仅使用SDK。

______________________________________________________________________

建筑原理

此实现演示了：

坚实的原则 --单一职责、开放/封闭、Liskov替换、接口隔离、依赖倒置
关注点分离 --层之间的清晰边界
依赖注入 --组件接收依赖关系，而不是创建依赖关系
DRY（不要重复自己） --将通用功能提取到基类中
代码配置 --所有可配置的值都在YAML中，不是硬编码的
可测试性 --每一层都可以独立测试，具有全面的测试覆盖率

建筑亮点

可替换性：

将STDIO交换为HTTP传输→ 只有传输层发生了变化
用Web UI替换CLI→ 仅UI层更改
添加新工具/资源/提示→ 仅MCP层发生变化

独立性：

每一层都对更高层一无所知
MCP服务器不知道传输机制
SDK不了解MCP服务器内部
CLI不知道传输或MCP

可扩展性：

通过实施添加新传输 BaseTransport
通过实施添加新工具 BaseTool
通过实施添加新资源 BaseResource
通过实现添加新提示 BasePrompt

______________________________________________________________________

贡献

欢迎投稿！拜托：

克隆该仓库
创建要素分支(git checkout -b feature/amazing-feature)
提交您的更改(git commit -m 'Add amazing feature')
推到分支(git push origin feature/amazing-feature)
打开拉取请求

代码质量标准

保持>70%的测试覆盖率
遵循PEP 8风格指南
将文件保持在150行以下
为所有函数添加类型提示
记录所有公共API

______________________________________________________________________

文档

其他文件可在 docs/ 目录：

架构文档 --带图表的详细架构概述
产品要求 --完整的需求规格说明
架构决策记录 --具有基本原理的关键架构决策
提示书 --人工智能辅助开发方法
研究范围 --评价方法

构建API文档

该项目包括使用Sphinx的自动化API文档。文档是从源代码中的文档字符串自动生成的。

先决条件：

pip install sphinx sphinx-rtd-theme

构建HTML文档：

sphinx-build -b html docs/ docs/_build/

查看文档：

open docs/_build/index.html  # macOS
xdg-open docs/_build/index.html  # Linux
start docs/_build/index.html  # Windows

生成的文档包括：

MCP服务器API（服务器、注册表、工具、资源、提示）
SDK客户端API（客户端操作和生命周期）
传输层API（基本传输、STDIO、处理程序）
核心基础设施API（配置、日志记录、错误）

______________________________________________________________________

许可证

该项目根据MIT许可证获得许可。看许可证了解详情。

______________________________________________________________________

作者

塔尔·巴达

github： @ 塔尔巴达8

______________________________________________________________________

致谢

内置：

python --核心语言和标准库
pytest --测试框架
格式 --配置管理
参数解析 --命令行界面

______________________________________________________________________

⭐ 如果你觉得这个项目有用，请考虑给它一颗星！

MCP Modular Architecture

快速接入

详细介绍

MCP模块化架构

目录

概述

什么是MCP？

这是给谁的？

建筑

层职责

1.核心基础设施

2.MCP服务器层

3.传输层

4.SDK层

5.用户界面

关键概念

工具

资源

提示

运输抽象

SDK优先设计

项目结构

安装

先决条件

设置

快速开始

使用模式

1.嵌入式服务器（库使用）

2.独立服务器+CLI

3.独立服务器+SDK

运行项目

配置

运行独立服务器

以编程方式使用MCP服务器（嵌入式）

CLI使用情况

可用命令

完整示例

SDK使用

连接到外部服务器

与现有服务器一起使用

SDK方法

运行测试

运行所有测试

跑步有保障

运行特定测试

检验统计量

保险范围详细信息

视觉示例

快速概览

1. SDK演示 -完成工作流程

2. 多处理 （CPU限制）

3. 多线程 （I/O绑定）

4. 完整测试套件

插件示例：系统可扩展性

WeatherTool插件

运行插件演示

关键要点

创建自己的插件

发展

插件开发

添加新工具

添加新传输

添加新资源

扩展CLI

建筑原理

建筑亮点

贡献

代码质量标准

文档

构建API文档

许可证

作者

致谢

目录标签

接入字段

权限和风险

来源信息

继续浏览同类 MCP

2. 多处理（CPU限制）

3. 多线程（I/O绑定）