MCP-Server

🧠 MCP 服务器工智能代理编排核心 [](https://python.org) [](https://.com) [ [](#-核心与代理通信协议) [ **MCP 服务器**Python 3** 的人工智能排核心，用于启动和协调人工智能代理。每个代理独立的子进程运行，通过简单的 JSON 协议与通信。核心存储共享内存，通过 **Ollama REST API** 访问，支持实时的交互式对话，并能自动运行用户请求的全流程，包括 **分析师 → 开发者 → 测试人员**，并在测试失败时自动修复代码。 --- ## 📋 目录 - [🚀 核心功能](#-核心功能) - [📂 项目结构](#-项目结构) - [🧩 已实现组件](#-已实现组件) - [🏗 架构](#-架构) - [🔌 核心与代理通信协议](#-核心通信协议) 🛠 要求](#-要求) - [⚙️ 安装](#️-安装) - [▶核心检查](#️-运行核心检查) - [💬 与代理交互式聊天](#-与代理交互式聊天) - [🔗 分析师 → 开发者 → 测试人员全流程](#-分析师-开发者-测试人员全流程) - [🗂 技术规格和文件系统安全性](#-技术规格和文件系统安全性) - [📚 Orchestrator 方法](#-orchestrator) - [👤 作者](#-作者) --- 🚀 核心功能 ### 🏗 架构 子进程代理**：每个代理都是一个单独（可以是任何语言，不仅仅是 Python），核心不关心其内部实现。 - **统一 JSON 协议**和代理之间的所有消息都是 JSON 字符串，通过 stdin/stdout 进行通信，不依赖于。 - **代理共享内存**：核心维护的简单键值存储在内存中，代理之间可以共享，通过自动保存到磁盘 ⚡ 非阻塞读取和对 Windows 编码的鲁棒性 - 每个代理的 **stdout** 和 **stderr** 读取都在单独中进行（ing.Thread` + `queue.Queue`），核心不会的代理而“冻结”，代理的诊断信息会立即在控制台中带有 `[agent:<名称>][stderr]` 前缀 代理子进程启动时强制设置 `PYTHONIOENCODING=utf-8` 和 `encoding="utf-8"`，避免在 Windows 上出现 `UnicodeEncodeError`（Windows 控制台过时的代码页，如 cp1251）。 🤖 与 LLM 集成 - 直接使用 **Ollama** 的 REST 客户端（`/api/generate`，`/api/tags`）而无需第三方 SDK，在核心和代理中均支持。 - 没有伪造的 LLM（FakeLLM）；如果 Ollama 不可用或超时，代理会返回清晰的错误消息。 - 强制使用 Ollama 的 JSON 模式（`"format": "json"`）进行任务请求，确保生成语法有效的 JSON。 ### 💬 实时对话与自动模式切换 - `run_analyst.py` 启动分析师代理并允许实时对话。 - 代码中实现意图检测器（`wants_spec()` in `run_analyst.py`）：如果消息中包含“任务要求” / “技术规格” / “项目计划”等词语，编排器会自动启用严格的验证模式。 ### 🔗 分析师 → 开发者 → 测试人员全流程 - 验证任务要求后，编排器 **自动** 启动开发者代理，传递任务要求并将生成的代码写入磁盘。 - 随后 **自动** 启动测试人员代理：为每个文件生成单元测试并运行 pytest。 - 如果测试失败，编排器会向开发者发送错误报告，请求在 **同一项目文件夹** 中修复代码，然后再次运行测试。循环重复直到成功或达到重试限制（默认为 2）。 ### 📝 可观察性 - 所有核心事件同时记录到控制台和按日期分割的文件中。 - 单独记录无效的 LLM 响应日志：`logs/invalid_llm_responses.log`。 - 单独记录文件操作日志：`logs/fs_operations.log`（写入内容、阻止内容、错误内容）。 - 单独记录 pytest 运行日志：`logs/test_runs.log`（测试通过/失败数、退出代码）。 --- ## 📂 项目结构 ```text mcp_server/ ├── core/ │ ├── __init__.py │ ├── orchestrator.py # Orchestrator 类 + AgentHandle + 构建、测试、修复循环 │ ├── parser.py # LLM 响应 JSON 验证器（任务要求模式） │ ├── fs_tools.py # 安全的文件写入/目录创建 │ └── mcp_tools.py # 运行 pytest + 解析结果 │ ├── agents/ │ ├── __init__.py │ ├── test_agent.py # 最小测试代理（回声协议） │ ├── analyst.py # 分析师代理：对话模式 + 任务要求生成 │ ├── developer.py # 开发者代理：任务要求 → 磁盘文件代码 │ └── tester.py # 测试人员代理：代码 → 单元测试 → pytest 运行 │ ├── workspace/ │ ├── configs/ │ │ └── config.json # Ollama 地址、代理注册、超时设置 │ ├── prompts/ │ │ ├── system_prompt.txt # 基本系统提示 │ │ ├── analyst.md # 分析师提示（对话 + 严格 JSON 模式） │ │ ├── developer.md # 开发者提示（仅代码，无 markdown） │ │ └── tester.md # 测试人员提示（pytest 代码） │ ├── rules/ │ │ └── rules.md # 核心与代理通信协议 │ └── projects/ # 开发者和测试人员写入文件 │ ├── logs/ # 核心日志、LLM 无效响应、文件操作日志、测试运行日志 ├── run_test.py # 使用测试代理检查核心 _analyst.py # 交互式聊天 + 自动运行全流程 └── requirements.txt ``` --- 🧩 已实现组件 ### ✅ 组件 1 — `Orchestrator` (`core/orchestrator.py`) 系统核心，负责代理的生命周期、内存和全流程。 - 启动/停止代理子进程（`start_agent`, ``, `stop_all），强制使用8 编码。 - 共享内存，持久化存储在 `workspace/memory.json`（`remember`, `recall`, `append_history`）。 - 接收用户输入并路由到链中的第一个代理（`handle_user_input`，可选 `parse_as_spec=True`）。 - 运行全流程（`_run_pipeline`）：构建 → 测试 → 修复 → 再次测试，带有重试限制。 ### ✅ 组件 2 — `AgentHandle` (`core/orchestrator.py`) 每个运行代理的内部包装。 - 为非阻塞读取 **stdout** 和 **stderr** 创建后台线程。 - 在读取线程和核心主循环之间使用消息队列（`queue.Queue`）。 - 正确终止进程，超时时回落到 `kill()`。 ### ✅ 组件 3 — `test_agent` (`agents/test_agent.py`) 最小测试代理，用于测试协议，回声响应。 ### ✅ 组件 4 — `analyst` (`agents/analyst.py`) 默认对话助理，按需生成结构化任务要求。 - 从 `workspace/prompts/analyst.md` 读取系统提示，从 `workspace/configs/config.json` 读取主机和模型。 - 当 `want_spec` 标志为真时，向 Ollama 强制使用 JSON 模式。 - **无伪造 LLM**：如果 Ollama 不可用，返回清晰的错误消息。 ### ✅ 组件 5 — `parser.py` (`core/parser.py`) 结构化 JSON 响应验证器。 - 从响应文本中提取 JSON 对象。 - **规范化字段名称**：如果模型将键翻译为中文，自动转换为预期模式。 - 验证必填字段和列表类型。 ### ✅ 组件 6 — `developer` (`agents/developer.py`) 开发者代理：将任务要求转换为实际代码文件，会根据测试人员报告进行迭代。 - 从消息或 `workspace/memory.json` 读取任务要求。 - 为每个文件生成上下文提示，请求 Ollama 生成代码。 - 支持 `fix_request`：重新生成代码文件，添加 pytest 输出和修复请求。 ### ✅ 组件 7 — `fs_tools.py` (`core/fs_tools.py`) 文件系统工具，确保安全写入。 - `write_file` 和 `create_directory` 仅在 `workspace/projects/` 内操作。 - 防止路径逃逸（`PathEscapeError`）。 ### ✅ 组件 8 — `tester` (`agents/tester.py`) 测试人员代理：为开发者代码生成单元测试并运行 pytest。 - 读取项目目录和已写入的文件列表。 - 为每个 `.py` 文件生成测试代码并运行 pytest。 ### ✅ 组件 9 — `mcp_tools.py` (`core/mcp_tools.py`) 测试运行工具。 - 查找虚拟环境中的 Python 解释器。 - 运行 pytest 并解析输出。 --- ## 🏗 架构 ### 数据流：用户 → 分析师 → 开发者 → 测试人员 → （修复） ``` ┌──────────────┐ handle_user_input() ┌────────────────┐ stdin (JSON) ┌──────────┐ │ 用户 │─────────────────────►│ Orchestrator │───────────────►│ analyst │ └──────────────┘ │ │ │(subprocess)│ │ + memory.json │◄───────────────┤ │ └───────┬────────┘ stdout (JSON) └──────────┘ │ 任务要求有效 ▼ ┌────────────────┐ build_request ┌──────────┐ │ _run_pipeline() │───────────────►│developer │ │ │◄───────────────┤ │ └───────┬────────┘ files_written └────┬─────┘ │ │ write_file() ▼ ▼ ┌────────────────┐ test_request workspace/projects/<slug>/ │ │───────────────►┌──────────┐ │ (同一循环) │◄───────────────┤ tester │ └───────┬────────┘ passed/failed └────┬─────┘ │ │ run_pytest() failed > 0 │ ▼ ▼ pytest 报告 fix_request → developer （重复，最多 2 次） ``` 依赖关系为 **用户 → 核心 → 代理**。所有三个代理都直接调用 Ollama；核心还可以通过 `query_ollama()` 直接调用 Ollama。 ### 为什么选择子进程而非线程/协程 | 标准 | 子进程代理 | 同一进程中的线程 | |---|---|---| | **故障隔离** | ✅ 代理崩溃不会影响核心 | ❌ 未处理的异常可能损坏进程 | | **实现语言** | ✅ 代理可以使用任何语言 | ❌ 仅限 Python | | **协议简单性** | ✅ JSON 字符串 — 通用接口 | ⚠️ 需要共享的 Python 对象 | | **并行性** | ✅ 真正的操作系统并行 | ⚠️ GIL 限制了纯 Python | --- ## 🔌 核心与代理通信协议 核心 → 代理（`stdin`）： ```json {"type": "user_input", "text": "为提醒机器人编写任务要求", "memory_snapshot": [...], "want_spec": true} ``` 分析师代理 → 核心（`stdout`），任务要求就绪： ```json {"type": "agent_response", "agent": "analyst", "text": "{\"title\": \"...\", \"files\": [...]}}", "source": "ollama"} ``` 核心 → 开发者代理（`stdin`），初始构建： ```json {"type": "build_request", "spec": {"title": "...", "files": [{"path": "main.py", "description": "..."}]}} ``` 核心 → 开发者代理（`stdin`），修复请求： ```json {"type": "fix_request", "spec": {...}, "project_dir": "bot_napominaniy", "test_report": {"passed": 1, "failed": 2, "raw_output": "..."}} ``` 开发者代理 → 核心（`stdout`）： ```json {"type": "agent_response", "agent": "developer", "project_dir": "bot_napominaniy", "files_written": ["bot_napominaniy/main.py"], "errors": []} → 测试人员代理（`stdin`）json {"type": "test_request", "project_dir": "bot_napominaniy", "files_written": [...], "spec": {...}} ``` 测试人员代理 → 核心（`stdout`）： ```json {"type": "agent_response", "agent": "tester", "project_dir": "bot_napominaniy", "tests_written": ["bot_napominaniy/tests/test_main.py"], "passed": 3, "failed": 1, "raw_output": "..."} ``` 代理终止： {"type": "shutdown"} ``` 完整 [`workspace/rules/rules.md`](workspace/rules/rules.md)。 --- ## 🛠 要求 - Python 3.10+ - [Ollama](https://ollama.com) — 已安装并本地运行（`ollama serve`） - 至少一个 Ollama 模型已下载（例如 `ollama pull llama3` 或 `ollama pull qwen2.5-coder:3b`） - `pytest`（通过 `requirements.txt` 自动安装 — 需要运行生成的项目测试） ⚙️ 安装 ```bash git clone https://github.com/yushmeh/MCP-Server cd MCP-Server pip install -r requirements.txt ``` Ollama 配置： ```bash ollama serve ollama pull llama3 ``` 在实际对话之前，建议“热身”模型： ```bash ollama run llama3 >>> 你好 ``` # дождитесь полного ответа, затем >>> /bye ``` --- ## ▶️ 启动内核测试 ```bash python run_test.py ``` 启动 `test_agent.py`，发送三个测试消息，显示总体内存并尝试连接到 Ollama。 --- ## 💬 与代理交互式聊天 ```bash python run_analyst.py ``` 一次性启动代理并开启实时对话。退出方式为 `exit`、`quit` 或 `退出`（或 `Ctrl+C`）。 ``` 代理已准备就绪。输入以下命令退出：exit / quit / 退出 您：你好 代理：您好！我能帮您什么？ 您：退出 ``` 如果 Ollama 不可用，代理将返回实际错误消息，对话可以继续。 --- ## 🔗 分析师 → 开发人员 → 测试人员链 如果消息中包含 **"需求"**、**"技术要求"** 或 **"项目计划"**，则 `run_analyst.py` 将（在代码级别，而非依赖模型）启用严格模式： ``` 您：为 Telegram 提醒服药机器人编制需求 🛠 为「...」启动开发人员代理... ✅ 在 workspace/projects/<slug>/ 中生成了 5 个文件 🧪 为「<slug>」启动测试人员代理... 测试结果：4 通过，1 失败 🔁 测试失败（1）。向开发者提交错误报告，请求「修复错误」（尝试 1/2）... ✅ 在 workspace/projects/<slug>/ 中生成了 5 个文件 🧪 为「<slug>」启动测试人员代理... 测试结果：5 通过，0 失败 ✅ 「<slug>」的所有测试均通过。任务完成。 ``` 幕后操作（方法 `Orchestrator._run_pipeline`）： 1. 指挥者请求 Ollama 使用 `"format": "json"`（强制 JSON）生成技术要求，`core/parser.py` 验证并规范响应； 2. 如果技术要求成功，保存到内存键 `last_tech_spec`，并启动 `developer.py` - 根据 `spec["files"]` 生成代码； 3. 立即启动 `tester.py` - 为每个文件生成测试并通过 `core/mcp_tools.run_pytest()` 运行； 4. 如果所有测试通过，管道以成功消息完成； 5. 如果有失败，指挥者向开发人员发送带有 pytest 输出的 `fix_request`，开发人员在**同一目录**下重新生成文件，测试人员重新运行测试； 6. 循环重复直到成功或达到 `max_fix_attempts`（默认为 2）- 然后管道以需要手动检查的消息正确停止，而不陷入循环； 7. 开发人员和测试人员的超时时间根据文件数量缩放 - 在没有 GPU 的情况下在 CPU 上生成可能需要每文件 1-2 分钟； 8. 如果在构建步骤发生超时，指挥者查看项目目录并显示已经写入的内容（代理进程不会被杀死，可以在后台继续写入文件）。 可以通过聊天方式调用链： ```python response = orch.handle_user_input("为 ... 编制技术要求", parse_as_spec=True) ``` --- ## 🗂 技术要求方案和文件系统安全性 分析师生成并由开发人员和测试人员使用的完整 JSON 方案： ```json { "title": "项目简称", "description": "任务的详细描述", "requirements": ["要求 1", "要求 2"], "modules": ["模块 1", "模块 2"], "tech_stack": ["技术 1", "技术 2"], "files": [ {"path": "main.py", "description": "应用程序入口点"}, {"path": "models.py", "description": "数据模型"} ] } ``` `title`、`description`、`requirements`、`modules` 为必填字段；`files` 为启动开发者时必填（否则使用备用 `main.py`）。 开发者和测试人员编写的所有文件都经过 `core/fs_tools.py`： * 解析并检查路径，确保严格位于 `workspace/projects/` 内； * 阻止绝对路径和 `../../` 越界访问（`PathEscapeError`）； * 每个操作都记录在 `logs/fs_operations.log` 中。 测试人员的测试额外通过 `core/mcp_tools.run_pytest()` 运行 - 在可用的最佳解释器（项目虚拟环境或 MCP 服务器环境）上，并记录在 `logs/test_runs.log` 中。 --- ## 📚 Orchestrator 方法 | 方法 | 用途 | |---|---| | `start_agent(name, script_path)` | 以子进程方式启动代理（UTF-8 编码） | | `stop_agent(name)` / `stop_all_agents()` | 停止一个或所有代理 | | `handle_user_input(text, parse_as_spec=False)` | 将输入传递给第一个代理；当 `True` 时，作为技术要求验证并启动完整管道 | | `remember(key, value)` / `recall(key)` | 代理内存的写入/读取 | | `append_history(role, content)` | 将记录追加到对话历史中 | | `query_ollama(prompt, model=...)` | 通过 Ollama（在内核级别）向本地 LLM 发送查询 | | `list_ollama_models()` | 已安装在 Ollama 中的模型列表 | | `_handle_analyst_response(response)` | 通过 `core.parser` 验证技术要求，保存到内存，启动 `_run_pipeline` | | `_run_pipeline(spec, max_fix_attempts=2)` | 完整周期：构建 → 测试 → 修复 → 测试，直到成功或尝试次数达到上限 | | `_ensure_agent_running(name, script)` | 按文件名启动/重用代理 | | `_trigger_developer(spec)` / `_trigger_developer_fix(...)` | 初始构建 / 带有错误报告的重复构建 | | `_trigger_tester(project_dir, files, spec)` | 为已构建的项目启动测试人员 | | `_print_dev_result(response)` | 打印构建结果（文件/错误） | | `_report_partial_project(spec)` | 超时时，显示已写入磁盘的内容 | | `shutdown()` | 正确停止所有代理并保存内存 | --- ## 👤 作者 [@yushmeh](https://github.com/yushmeh)