WeChat Article Reader MCP Server

# 开发指南 本文档为微信公众号文章读取 MCP 服务器的开发者提供详细的开发指南。 ## 项目结构 ``` mcp-server-wechat/ ├── src/ │ └── mcp_server_wechat/ │ ├── __init__.py │ ├── server.py # MCP 服务器主入口 │ ├── http_server.py # HTTP 传输协议实现 │ ├── models/ │ │ ├── __init__.py │ │ ├── article.py # 文章数据模型 │ │ └── response.py # 响应数据模型 │ ├── tools/ │ │ ├── __init__.py │ │ ├── base.py # 工具基类 │ │ ├── fetch_article.py # 获取文章工具 │ │ ├── search_articles.py # 搜索文章工具 │ │ ├── extract_summary.py # 提取摘要工具 │ │ └── wechat_tools.py # MCP 工具注册与处理 │ └── utils/ │ ├── __init__.py │ ├── browser.py # 浏览器工具 │ ├── parser.py # 内容解析器 │ ├── formatter.py # 响应格式化器 │ └── validators.py # 输入验证器 ├── tests/ │ ├── __init__.py │ ├── test_wechat_tools.py # 单元测试 │ ├── test_integration.py # 集成测试 │ └── test_performance.py # 性能测试 ├── docs/ │ ├── API.md # API 文档 │ ├── DEPLOYMENT.md # 部署指南 │ └── DEVELOPMENT.md # 开发指南 ├── requirements.txt # 依赖列表 ├── setup.py # 安装脚本 ├── run_tests.py # 测试运行器 └── README.md # 项目说明 ``` ## 开发环境设置 ### 1. 克隆仓库 ```bash git clone <repository-url> cd mcp-server-wechat ``` ### 2. 创建虚拟环境 ```bash python -m venv venv source venv/bin/activate # Linux/macOS # 或 venv\Scripts\activate # Windows ``` ### 3. 安装开发依赖 ```bash pip install -r requirements.txt pip install -r requirements-dev.txt # 如果存在 ``` ### 4. 安装预提交钩子 ```bash pre-commit install ``` ## 核心组件 ### 1. MCP 服务器 MCP 服务器是项目的核心组件，负责处理客户端请求并调用相应的工具。 主要文件：`src/mcp_server_wechat/server.py` 关键功能： - 初始化 MCP 服务器 - 注册工具 - 处理工具调用请求 - 格式化响应 ### 2. 工具系统 工具是 MCP 服务器的功能单元，每个工具负责一个特定的任务。 #### 基类 所有工具都继承自 `BaseTool` 类： ```python class BaseTool(ABC): @abstractmethod async def execute(self, **kwargs) -> Dict[str, Any]: """执行工具逻辑""" pass @abstractmethod def get_schema(self) -> Dict[str, Any]: """获取工具的 JSON Schema""" pass ``` #### 实现新工具 1. 创建工具类，继承 `BaseTool` 2. 实现 `execute` 方法 3. 实现 `get_schema` 方法 4. 在 `wechat_tools.py` 中注册工具 示例： ```python class MyCustomTool(BaseTool): async def execute(self, param1: str, param2: int = 10) -> Dict[str, Any]: # 实现工具逻辑 result = do_something(param1, param2) return { "success": True, "message": "操作成功", "data": result } def get_schema(self) -> Dict[str, Any]: return { "name": "my_custom_tool", "description": "我的自定义工具", "inputSchema": { "type": "object", "properties": { "param1": { "type": "string", "description": "参数1的描述" }, "param2": { "type": "integer", "description": "参数2的描述", "default": 10 } }, "required": ["param1"] } } ``` ### 3. 数据模型 数据模型定义了系统中使用的数据结构。 #### 文章模型 ```python @dataclass class WeChatArticle: url: str title: str author: str publish_time: str account_name: str account_id: str read_count: int like_count: int comment_count: int extracted_at: str ``` #### 响应模型 ```python @dataclass class ResponseMetadata: success: bool message: str request_id: Optional[str] = None error_code: Optional[str] = None error_type: Optional[str] = None ``` ### 4. 工具函数 工具函数提供各种辅助功能。 #### 浏览器工具 `src/mcp_server_wechat/utils/browser.py` 提供浏览器操作功能： ```python async def get_browser_client(headless: bool = True) -> Browser: """获取浏览器客户端""" async def fetch_page_content(url: str, timeout: int = 30) -> str: """获取页面内容""" ``` #### 内容解析器 `src/mcp_server_wechat/utils/parser.py` 提供内容解析功能： ```python class WeChatArticleParser: def parse_article(self, html: str) -> WeChatArticle: """解析文章内容""" def extract_text_content(self, html: str) -> str: """提取文本内容""" ``` #### 响应格式化器 `src/mcp_server_wechat/utils/formatter.py` 提供响应格式化功能： ```python def format_mcp_tool_response( success: bool, message: str, data: Optional[Dict[str, Any]] = None, request_id: Optional[str] = None, error_code: Optional[str] = None, error_type: Optional[str] = None, format_type: str = "json" ) -> str: """格式化 MCP 工具响应""" ``` ## 测试 ### 运行测试 ```bash # 运行所有测试 python run_tests.py # 只运行单元测试 python run_tests.py --unit # 只运行集成测试 python run_tests.py --integration # 只运行性能测试 python run_tests.py --performance ``` ### 编写测试 #### 单元测试 ```python import unittest from src.mcp_server_wechat.tools.fetch_article import FetchArticleTool class TestFetchArticleTool(unittest.TestCase): def setUp(self): self.tool = FetchArticleTool() async def test_execute_success(self): # 测试成功情况 result = await self.tool.execute(url="valid_url") self.assertTrue(result["success"]) async def test_execute_invalid_url(self): # 测试无效 URL result = await self.tool.execute(url="invalid_url") self.assertFalse(result["success"]) self.assertEqual(result["error_code"], "INVALID_URL") ``` #### 集成测试 ```python import unittest from src.mcp_server_wechat.tools.wechat_tools import handle_call_tool class TestIntegration(unittest.TestCase): async def test_fetch_article_integration(self): request = { "name": "fetch_article", "arguments": { "url": "https://mp.weixin.qq.com/s/xxxxxxxx" } } response = await handle_call_tool(request) self.assertTrue(response["success"]) ``` ## 代码规范 ### 1. Python 代码风格 项目遵循 PEP 8 代码风格指南。使用以下工具确保代码质量： - `black`: 代码格式化 - `flake8`: 代码检查 - `isort`: 导入排序 - `mypy`: 类型检查 ### 2. 文档字符串 所有公共函数、方法和类都应有详细的文档字符串： ```python def fetch_article(url: str, content_formats: List[str] = None) -> Dict[str, Any]: """ 获取微信公众号文章内容 Args: url: 文章URL content_formats: 内容格式列表，可选值：markdown, text Returns: 包含文章内容和元数据的字典 Raises: ValidationError: 当URL无效时 NetworkError: 当网络请求失败时 """ ``` ### 3. 类型注解 所有函数和方法都应有类型注解： ```python def process_data( input_data: Dict[str, Any], config: Optional[Dict[str, Any]] = None ) -> Tuple[bool, str]: """处理输入数据""" ``` ## 调试 ### 1. 日志记录 项目使用 Python 标准库的 `logging` 模块： ```python import logging logger = logging.getLogger(__name__) async def fetch_article(url: str) -> Dict[str, Any]: logger.info(f"开始获取文章: {url}") try: # 执行操作 result = await do_something(url) logger.info(f"文章获取成功: {url}") return result except Exception as e: logger.error(f"文章获取失败: {url}, 错误: {str(e)}") raise ``` ### 2. 调试技巧 1. **使用断点** ```python import pdb; pdb.set_trace() ``` 2. **打印变量** ```python print(f"变量值: {variable}") ``` 3. **使用日志** ```python logger.debug(f"调试信息: {variable}") ``` ## 性能优化 ### 1. 异步编程 项目大量使用异步编程模式，确保高并发性能： ```python async def fetch_multiple_articles(urls: List[str]) -> List[Dict[str, Any]]: """并发获取多篇文章""" tasks = [fetch_article(url) for url in urls] results = await asyncio.gather(*tasks, return_exceptions=True) return results ``` ### 2. 缓存 实现缓存机制，减少重复请求： ```python from functools import lru_cache @lru_cache(maxsize=128) def get_article_metadata(url: str) -> Dict[str, Any]: """获取文章元数据（带缓存）""" # 实现逻辑 ``` ### 3. 资源管理 合理管理浏览器等资源： ```python async def with_browser(): browser = await get_browser_client() try: # 使用浏览器 result = await do_something_with_browser(browser) return result finally: await browser.close() ``` ## 发布流程 ### 1. 版本控制 项目使用语义化版本控制（Semantic Versioning）： - 主版本号：不兼容的 API 修改 - 次版本号：向下兼容的功能性新增 - 修订号：向下兼容的问题修正 ### 2. 发布步骤 1. 更新版本号 2. 更新 CHANGELOG.md 3. 运行完整测试套件 4. 创建 Git 标签 5. 构建分发包 6. 发布到 PyPI ```bash # 更新版本号 bump2version patch # 或 minor, major # 运行测试 python run_tests.py # 创建标签 git tag -a v1.0.0 -m "Release version 1.0.0" git push origin v1.0.0 # 构建分发包 python setup.py sdist bdist_wheel # 发布到 PyPI twine upload dist/* ``` ## 贡献指南 ### 1. 提交代码 1. Fork 项目 2. 创建功能分支 3. 提交更改 4. 推送到分支 5. 创建 Pull Request ```bash # 创建功能分支 git checkout -b feature/new-feature # 提交更改 git add . git commit -m "feat: 添加新功能" # 推送到分支 git push origin feature/new-feature ``` ### 2. 提交信息规范 使用 [Conventional Commits](https://www.conventionalcommits.org/) 规范： - `feat`: 新功能 - `fix`: 修复 bug - `docs`: 文档更新 - `style`: 代码格式化 - `refactor`: 代码重构 - `test`: 测试相关 - `chore`: 构建过程或辅助工具的变动 示例： ``` feat: 添加文章搜索功能 - 实现基于关键词的文章搜索 - 添加排序和过滤选项 - 更新 API 文档 ``` ## 常见问题 ### 1. 浏览器启动失败 确保系统安装了 Chrome 或 Chromium，并且版本与 ChromeDriver 兼容。 ### 2. 请求超时 调整 `BROWSER_TIMEOUT` 环境变量值，或检查网络连接。 ### 3. 内存不足 减少 `MAX_CONCURRENT_REQUESTS` 环境变量值，或增加服务器内存。 ### 4. 测试失败 确保所有依赖已正确安装，并且环境变量已正确设置。 ## 扩展指南 ### 1. 添加新的工具 1. 在 `src/mcp_server_wechat/tools/` 目录下创建新工具文件 2. 继承 `BaseTool` 类并实现必要方法 3. 在 `wechat_tools.py` 中注册新工具 4. 添加相应的测试 ### 2. 支持新的内容格式 1. 在 `WeChatArticleParser` 类中添加新的解析方法 2. 更新 `fetch_article` 工具以支持新格式 3. 更新 API 文档 ### 3. 添加新的传输协议 1. 在 `src/mcp_server_wechat/` 目录下创建新的传输协议文件 2. 实现必要的协议处理逻辑 3. 更新服务器入口点以支持新协议 4. 更新部署文档