backend-integrator

Backend 集成者

此技能提供将新的大模型后端集成到 MassGen 所需的完整检查清单和模式。完整集成需要修改代码库中约 15 个文件。

何时使用此技能

• 添加新的大模型提供商／后端
• 审查现有后端是否缺少集成环节
• 了解扩展后端功能时需要修改哪些文件

集成架构

Backend Type Decision:
  Stateless + OpenAI-compatible API   → subclass ChatCompletionsBackend
  Stateless + custom API              → subclass CustomToolAndMCPBackend
  Stateless + Response API format     → subclass ResponseBackend
  Stateful CLI wrapper (like Codex, Gemini CLI) → subclass LLMBackend directly
  Stateful SDK wrapper (like Claude Code, Copilot) → subclass LLMBackend directly

完整检查清单

阶段 1：核心实现（3 个文件）

1.1 后端类

文件：massgen/backend/<name>.py

选择基类：

• LLMBackend — 最精简的基类，所有处理都由你负责
• CustomToolAndMCPBackend — 添加 MCP 和自定义工具支持（最常用）
• ChatCompletionsBackend — 用于兼容 OpenAI 的 API（继承自上面的基类）
• ResponseBackend — 用于 OpenAI Response API 格式

必需方法：

async def stream_with_tools(self, messages, tools, **kwargs) -> AsyncGenerator[StreamChunk, None]:
    """Main streaming method. Yield StreamChunks."""

def get_provider_name(self) -> str:
    """Return provider name string (e.g., 'OpenAI', 'Codex')."""

def get_filesystem_support(self) -> FilesystemSupport:
    """Return NONE, NATIVE, or MCP."""

要生成的 StreamChunk 类型：

类型	适用时机	关键字段
"content"	文本输出	content="..."
"tool_calls"	调用工具	tool_calls=[{id, name, arguments}]
"reasoning"	思考／推理增量	reasoning_delta="..."
"reasoning_done"	推理完成	reasoning_text="..."
"reasoning_summary"	推理摘要增量	reasoning_summary_delta="..."
"reasoning_summary_done"	推理摘要完成	reasoning_summary_text="..."
"complete_message"	完整助手消息	complete_message={...}
"complete_response"	原始 API 响应	response={...}
"done"	流传输完成	usage={prompt_tokens, completion_tokens, total_tokens}
"error"	发生错误	error="..."
"agent_status"	状态更新	status="...", detail="..."
"backend_status"	后端级状态	status="...", detail="..."
"compression_status"	压缩事件	status="...", detail="..."
"hook_execution"	Hook 已运行	hook_info={...}, tool_call_id="..."

所有块的常见字段：source（Agent／编排器 ID）、display（布尔值，默认为 True）。

Token 跟踪 — 调用以下其中一项：

self._update_token_usage_from_api_response(usage_dict, model)  # If API returns usage
self._estimate_token_usage(messages, response_text, model)      # Fallback

计时 — 在 stream_with_tools 中调用：

self.start_api_call_timing(self.model)       # Before API call
self.record_first_token()                     # On first content chunk
self.end_api_call_timing(success=True/False)  # After completion

对于有状态后端（CLI／SDK 封装器），还需实现：

def is_stateful(self) -> bool: return True
async def clear_history(self) -> None: ...
async def reset_state(self) -> None: ...

压缩支持 — 继承 StreamingBufferMixin 并调用：

self._clear_streaming_buffer(**kwargs)       # Start of stream
self._finalize_streaming_buffer(agent_id=id) # End of stream

1.2 格式化器（如需要）

文件：massgen/formatter/<name>_formatter.py

仅当 API 使用非标准消息／工具格式（而非 OpenAI 聊天补全格式）时才需要。继承 FormatterBase 并实现 format_messages()、format_tools()、format_mcp_tools()。

现有格式化器：

• _claude_formatter.py — Anthropic Messages API
• _gemini_formatter.py — Gemini API
• _chat_completions_formatter.py — OpenAI／通用格式（兼容的 API 可复用）
• _response_formatter.py — OpenAI Response API 格式

1.3 API 参数处理器（如需要）

文件：massgen/api_params_handler/<name>_api_params_handler.py

仅当后端调用 HTTP API，且需要在将 YAML 配置参数传给 API 前对其进行筛选／转换时才需要。继承 APIParamsHandlerBase。

CLI／SDK 封装器（Codex、Claude Code）通常不需要此处理器 — 它们会直接构建命令。

阶段 2：注册（4 个文件）

2.1 后端 init.py

文件：massgen/backend/__init__.py

from .your_backend import YourBackend
# Add to __all__

2.2 CLI 后端映射

文件：massgen/cli.py

将以下内容添加到 create_backend() 函数：

elif backend_type == "your_backend":
    api_key = kwargs.get("api_key") or os.getenv("YOUR_API_KEY")
    if not api_key:
        raise ConfigurationError(
            _api_key_error_message("YourBackend", "YOUR_API_KEY", config_path)
        )
    return YourBackend(api_key=api_key, **kwargs)

对于不需要 API 密钥的 CLI 后端，跳过密钥检查。

2.3 能力注册表

文件：massgen/backend/capabilities.py

在 BACKEND_CAPABILITIES 中添加条目：

"your_backend": BackendCapabilities(
    backend_type="your_backend",
    provider_name="YourProvider",
    supported_capabilities={"mcp", "web_search", ...},
    builtin_tools=["web_search"],  # Provider-native tools
    filesystem_support="mcp",      # "none", "mcp", or "native"
    models=["model-a", "model-b"], # Newest first
    default_model="model-a",
    env_var="YOUR_API_KEY",        # Or None
    notes="...",
    model_release_dates={"model-a": "2025-06"},
    base_url="https://api.example.com/v1",  # If applicable
)

2.4 配置验证器（如需要）

文件：massgen/config_validator.py

如果有特殊规则（例如必需参数、不兼容的组合），则在 _validate_backend() 中添加特定于后端的验证。

阶段 3：Token 管理（1 个文件）

3.1 定价

文件：massgen/token_manager/token_manager.py

先检查 LiteLLM——仅当模型不在 LiteLLM 数据库中时，才将其添加到 PROVIDER_PRICING。提供商名称必须与 get_provider_name() 完全一致（区分大小写）。

第 4 阶段：排除参数（2 个文件，如果添加新的 YAML 参数）

4.1 基类排除项

文件：massgen/backend/base.py -> get_base_excluded_config_params()

4.2 API 参数处理器排除项

文件：massgen/api_params_handler/_api_params_handler_base.py -> get_base_excluded_params()

两者必须保持同步。添加任何不应传递给提供商 API 的框架级 YAML 参数。

第 5 阶段：身份验证

5.1 API 密钥后端（标准方式）

大多数后端使用 API 密钥。在 capabilities.py 中设置 env_var，并在 cli.py 中添加密钥检查：

# cli.py
api_key = kwargs.get("api_key") or os.getenv("YOUR_API_KEY")
if not api_key:
    raise ConfigurationError(...)
return YourBackend(api_key=api_key, **kwargs)

5.2 OAuth / 订阅身份验证（CLI/SDK 封装器）

对于支持 OAuth 的后端（如 Codex、Claude Code），实现多层身份验证级联：

def __init__(self, api_key=None, **kwargs):
    # Tier 1: Explicit API key
    self.api_key = api_key or os.getenv("YOUR_API_KEY")
    # Tier 2: Check cached OAuth tokens
    self.use_oauth = not bool(self.api_key)
    self.auth_file = Path.home() / ".your_cli" / "auth.json"

async def _ensure_authenticated(self):
    if self.api_key:
        os.environ["YOUR_API_KEY"] = self.api_key
        return
    if self._has_cached_credentials():
        return
    # Tier 3: Initiate OAuth flow
    await self._initiate_oauth_flow()

async def _initiate_oauth_flow(self, use_device_flow=False):
    """Wrap the CLI's login command."""
    cmd = [self._cli_path, "login"]
    if use_device_flow:
        cmd.append("--device-auth")  # For headless/SSH
    proc = await asyncio.create_subprocess_exec(*cmd, ...)

在 cli.py 中：跳过 API 密钥检查，将身份验证交由后端处理：

elif backend_type == "your_backend":
    # Auth handled by backend (API key or OAuth)
    return YourBackend(**kwargs)

在 capabilities.py 中：将 env_var 设置为 API 密钥名称（可选，OAuth 不要求）：

env_var="YOUR_API_KEY",  # Optional - can use OAuth instead
notes="Works with subscription auth (via `your-cli login`) or YOUR_API_KEY."

注意：我们仅通过受支持的 SDK 或提供商建议的程序化命令行用法提供 OAuth。我们必须确认每个后端接受的 OAuth 用法，以确保支持方式正确（例如，使用 Claude Agent SDK，而非通过 API 伪装使用 Claude Code）。

5.3 无身份验证（本地推理）

对于本地服务（LM Studio、vLLM、SGLang），在 capabilities 中设置 env_var=None。无需在 cli.py 中检查密钥。

参考实现：

• codex.py — 完整的 OAuth，支持浏览器和设备代码流程
• claude_code.py — 3 层身份验证：CLAUDE_CODE_API_KEY -> ANTHROPIC_API_KEY -> 订阅登录
• lmstudio / vllm / sglang — 无身份验证（env_var=None）

第 6 阶段：自定义工具与 MCP

6.1 标准路径（API 后端）

如果继承自 CustomToolAndMCPBackend，MCP 和自定义工具会自动生效：

• 来自 YAML mcp_servers 配置的 MCP 服务
• 来自 custom_tools / custom_tools_path 配置的自定义工具
• 设置 cwd 时，由 FilesystemManager 注入文件系统 MCP
• 由 ToolManager 处理工具执行

6.2 多模态工具（所有后端类型）

设置 enable_multimodal_tools: true 时，后端必须注册 read_media 和 generate_media 自定义工具。使用 massgen/backend/base.py 中的共享辅助函数：

from .base import get_multimodal_tool_definitions

# In __init__:
enable_multimodal = self.config.get("enable_multimodal_tools", False) or kwargs.get("enable_multimodal_tools", False)
if enable_multimodal:
    custom_tools.extend(get_multimodal_tool_definitions())

• API 后端（CustomToolAndMCPBackend 子类）：自动处理——基类调用 get_multimodal_tool_definitions()，并通过 _register_custom_tools() 注册
• CLI/SDK 后端（Codex、Claude Code 等 LLMBackend 子类）：必须在 __init__ 中显式执行此操作，然后封装为 MCP 服务（见 §6.4）

重要：始终使用 get_multimodal_tool_definitions()——切勿内联工具字典。这样可以将定义集中在一处。

6.3 CLI/SDK 封装器——MCP 服务

这些后端必须配置 CLI/SDK 自带的 MCP 系统：

Codex：在工作区中写入项目范围的 .codex/config.toml（-C 目录）。Codex 会自动读取此文件。将 MassGen 的 mcp_servers 列表转换为 TOML 格式：

[mcp_servers.filesystem]
command = "npx"
args = ["-y", "@modelcontextprotocol/server-filesystem", "/workspace"]

Claude Code：将 mcp_servers 字典直接传递给 SDK 选项。SDK 会处理服务生命周期：

options = {"mcp_servers": {"filesystem": {"command": "npx", "args": [...]}}}

6.4 CLI/SDK 包装器 — 自定义工具

自定义工具需要特殊处理，因为大模型运行在外部进程中：

首选：包装为 MCP 服务（claude_code.py 采用此方式）：

1. 通过 ToolManager 加载自定义工具（架构 + 执行器）
2. 创建一个 MCP 服务，公开每个工具
3. 向 CLI/SDK 注册 MCP 服务
4. 大模型调用 MCP 工具时，包装器通过 ToolManager 执行该工具

# claude_code.py pattern (simplified)
def _create_sdk_mcp_server_from_custom_tools(self):
    tool_schemas = self._custom_tool_manager.fetch_tool_schemas()
    mcp_tools = []
    for schema in tool_schemas:
        name = schema["function"]["name"]
        async def wrapper(args, tool_name=name):
            return await self._execute_massgen_custom_tool(tool_name, args)
        mcp_tools.append(tool(name=name, ...)(wrapper))
    return create_sdk_mcp_server(name="massgen_custom_tools", tools=mcp_tools)

对于不支持 SDK MCP 的 CLI 包装器（如 Codex）： 使用共享的 massgen/mcp_tools/custom_tools_server.py 工具。它会创建一个独立的 fastmcp 服务，通过 stdio 传输方式包装 ToolManager 工具。

from ..mcp_tools.custom_tools_server import build_server_config, write_tool_specs

def _setup_custom_tools_mcp(self, custom_tools):
    # 1. Write tool specs JSON to workspace
    specs_path = Path(self.cwd) / ".codex" / "custom_tool_specs.json"
    write_tool_specs(custom_tools, specs_path)

    # 2. Get MCP server config (fastmcp run ... --tool-specs ...)
    server_config = build_server_config(
        tool_specs_path=specs_path,
        allowed_paths=[self.cwd],
        agent_id="my_backend",
    )

    # 3. Add to mcp_servers list (written to workspace config before launch)
    self.mcp_servers.append(server_config)

CLI 会将该服务作为子进程启动，并通过 stdio 连接。请在 reset_state() 时清理 specs 文件。

参考：codex.py 采用了此模式。custom_tools_server.py 还提供了 build_server_config()，可返回一个可直接使用的 MCP 服务字典。

备用方案：注入系统提示词

• 在系统提示词中将工具描述为 JSON 架构
• 解析结构化输出（JSON 块）以获取工具调用
• 仅在无法通过 MCP 包装时使用

6.5 自定义工具 YAML 配置

backend:
  type: your_backend
  custom_tools:
    - path: "massgen/tool/_basic"          # Directory with TOOL.md
      function: "two_num_tool"
    - path: "path/to/tool.py"             # Single file
      function: ["func_a", "func_b"]
      preset_args: [{"timeout": 30}, {}]
  custom_tools_path: "massgen/tool/"      # Auto-discover from directory
  auto_discover_custom_tools: true        # Discover from registry

MassGen 工作流工具（new_answer、vote 等）：始终通过系统提示词注入——这些工具属于协调层，而非可执行工具。完整详情请参阅 §6.7。

6.7 工作流工具集成（vote、new_answer 等）

工作流工具不是原生函数调用工具——它们会注入系统提示词，并从模型的文本输出中解析。Claude Code 和 Codex 后端采用了这一共同模式。

massgen/backend/base.py 中的共享辅助函数：

from .base import build_workflow_instructions, parse_workflow_tool_calls, extract_structured_response

# build_workflow_instructions(tools) → str
#   Filters tools to workflow tools, returns instruction text with usage examples.
#   Returns "" if no workflow tools present.

# parse_workflow_tool_calls(text) → List[Dict]
#   Extracts JSON tool calls from text output.
#   Returns standard format: {id, type, function: {name, arguments}}

# extract_structured_response(text) → Optional[Dict]
#   Low-level: extracts {"tool_name": "...", "arguments": {...}} from text.
#   Tries: ```json blocks → regex → brace-matching → line-by-line.

对于 API 后端：工作流工具会作为普通函数工具传入——无需注入。

对于 CLI/SDK 后端（Codex、Claude Code）：模型无法接收原生函数工具定义，因此：

1. 构建指令：调用 build_workflow_instructions(tools) 获取指令文本
2. 注入系统提示词：将其追加到后端用于系统提示词的相应机制中
3. 累积文本输出：在流式传输期间跟踪所有 content 块
4. 流式传输后解析：调用 parse_workflow_tool_calls(accumulated_text) 提取工具调用
5. 作为完成块生成：yield StreamChunk(type="done", tool_calls=workflow_tool_calls)

Codex 专用：指令会写入工作区根目录下的 AGENTS.md（Codex 会自动读取此文件）。同时也会从消息中提取编排器的系统消息并将其纳入其中，因为 Codex 通过 CLI 只能接收单条用户提示词——否则编排器的系统消息会丢失。采用此方法是因为开发者指令方法无法正常工作。

Claude Code 专用：指令由 _build_system_prompt_with_workflow_tools() 构建，该函数还会添加工具调用部分，然后作为 system_prompt 传递给 SDK。

典型的工作流工具解析方式（例如 Claude Code）是否可行，还是需要使用 MCP 工具（例如 Codex），目前尚无定论。

6.7.1 基于 MCP 与基于文本的工作流工具

为 CLI/SDK 后端选择基于 MCP 还是基于文本的工作流工具时：

基于 MCP 的方案（已尝试，但并非总是可靠）：

• 将工作流工具注册为 MCP 服务
• 将其传递给 SDK/CLI 的 MCP 系统
• 注意：MCP 工具的命名方式因 SDK 而异。Claude Code 会添加前缀 mcp__{server_name}__：

  # If server is "massgen_workflow_tools", tool names become:
  # "new_answer" → "mcp__massgen_workflow_tools__new_answer"
  # "vote" → "mcp__massgen_workflow_tools__vote"
  

• 使用 MCP 工作流工具时，请使用带有相应前缀的 build_workflow_mcp_instructions()：

  instructions = build_workflow_mcp_instructions(
      tools,
      mcp_prefix="mcp__massgen_workflow_tools__"
  )
  

基于文本的方案（Claude Code 当前的备用方案）：

• 将工作流工具指令注入系统提示词
• 使用 parse_workflow_tool_calls() 从模型的文本输出中解析 JSON 工具调用
• 更可靠，但需要健壮的解析机制

当前状态：由于基于 MCP 的方案不可靠，Claude Code 使用基于文本的工作流工具（JSON 解析）。Codex 也通过 AGENTS.md 指令使用基于文本的方案。

6.6 提供方原生工具：保留还是覆盖

基于 CLI/SDK 的 Agent（Codex、Claude Code）自带内置工具（文件编辑、Shell 执行、网页搜索、子 Agent 等）。集成时，请决定哪些工具要保留，哪些要用 MassGen 的等效工具覆盖。在 NativeToolBackendMixin 的 get_tool_category_overrides() 中记录这些决定——详情请参阅 §6.8 和阶段 10.2。

一般规则：优先使用提供方的原生工具，除非 MassGen 有特定原因需要覆盖。原生工具针对提供方的模型进行了优化，也能避免额外的 MCP 开销。

在 MassGen 能带来价值时进行覆盖：

• 子 Agent/任务工具：用 MassGen 的子 Agent 覆盖——提供方的子 Agent 无法参与 MassGen 协调（投票、共识、情报共享）
• 文件系统工具：如果 MassGen 需要执行路径权限控制或工作区隔离，且提供方不支持，可考虑覆盖

保留提供方原生工具：

• 文件读取/写入/编辑：提供方工具经过模型优化
• Shell/命令执行：提供方的沙箱已配置完成
• 网页搜索：提供方集成通常十分顺畅

实现：使用提供方的工具筛选机制禁用特定工具：

# Codex: via .codex/config.toml
[mcp_servers.some_server]
disabled_tools = ["task", "sub_agent"]

# Claude Code: via SDK disallowed_tools param
all_params["disallowed_tools"] = [
    "Read", "Write", "Edit", "MultiEdit",  # Use MassGen's MCP filesystem
    "Bash", "BashOutput", "KillShell",     # Use MassGen's execute_command
    "LS", "Grep", "Glob",                  # Use MassGen's filesystem tools
    "TodoWrite",                            # MassGen has its own task tracking
    "NotebookEdit", "NotebookRead",
    "ExitPlanMode",
    # Security restrictions
    "Bash(rm*)", "Bash(sudo*)", "Bash(su*)", "Bash(chmod*)", "Bash(chown*)",
]
# Conditionally keep web tools:
if not enable_web_search:
    disallowed_tools.extend(["WebSearch", "WebFetch"])

关键模式：

• Claude Code SDK 支持使用 glob 模式限制 Bash："Bash(rm*)"、"Bash(sudo*)"
• 检查 if "disallowed_tools" not in all_params，以允许用户通过 YAML 配置覆盖
• Codex 在 .codex/config.toml 中为每个 MCP 服务使用 disabled_tools；对于内置工具，Codex 没有禁用机制——改用 --full-auto 自动批准

在 capabilities.py 中记录后端提供的原生工具：

builtin_tools=["file_edit", "shell", "web_search", "sub_agent"],
notes="Native tools: file ops, shell, web search, sub-agents. Override sub-agents with MassGen's."

6.8 NativeToolBackendMixin

文件：massgen/backend/native_tool_mixin.py

对于带有内置工具的后端（CLI/SDK 封装），使用 NativeToolBackendMixin 统一工具筛选和钩子集成：

from .native_tool_mixin import NativeToolBackendMixin

class YourBackend(NativeToolBackendMixin, LLMBackend):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.__init_native_tool_mixin__()
        # Optional: initialize hook adapter
        self._init_native_hook_adapter(
            "massgen.mcp_tools.native_hook_adapters.YourBackendAdapter"
        )

    def get_disallowed_tools(self, config):
        """Return native tools to disable (MassGen has equivalents)."""
        return ["NativeTool1", "NativeTool2"]

混入类提供：

方法	用途
get_disallowed_tools(config)	抽象方法 — 声明要禁用的原生工具
get_tool_category_overrides()	抽象方法 — 声明要跳过/覆盖的 MCP 类别
supports_native_hooks()	检查钩子适配器是否可用
get_native_hook_adapter()	获取适配器实例
set_native_hooks_config(config)	设置原生格式的 MassGen 钩子
_init_native_hook_adapter(path)	按导入路径初始化适配器

参考实现：

• claude_code.py — 禁用大多数原生工具（Read、Write、Bash 等），改用 MassGen MCP
• codex.py — 保留所有原生工具（MassGen 会改为跳过附加等效的 MCP 工具）

阶段 7：测试（2 个以上文件）

7.1 能力测试

在 capabilities.py 中添加后，massgen/tests/test_backend_capabilities.py 会自动测试。

uv run pytest massgen/tests/test_backend_capabilities.py -v

7.2 集成测试

创建：massgen/tests/test_<name>_integration.py，或将测试添加到跨后端测试脚本中。

跨后端测试模式：

BACKEND_CONFIGS = {
    "claude": {"type": "claude", "model": "claude-haiku-4-5-20251001"},
    "openai": {"type": "openai", "model": "gpt-4o-mini"},
    "your_backend": {"type": "your_backend", "model": "model-a"},
}

7.3 钩子触发测试（CLI/SDK 后端）⚠️ 必需

脚本：scripts/test_hook_backends.py

**每个新的 CLI/SDK 后端都必须进行此测试。**该脚本会验证在真实流式处理期间，完整的钩子流水线是否确实触发，而不仅仅是钩子数据结构是否正确。

两种模式：

单元模式（不调用 API — 在 CI 中运行）：

uv run python scripts/test_hook_backends.py --backend your_backend

验证：后端存储了 GeneralHookManager，MidStreamInjectionHook 返回注入内容，HighPriorityTaskReminderHook 触发，组合钩子能够正确聚合。

E2E 模式（调用真实 API）：

uv run python scripts/test_hook_backends.py --backend your_backend --e2e
uv run python scripts/test_hook_backends.py --backend your_backend --e2e --verbose

验证完整的实时流程：PreToolUse 钩子触发 → 工具执行 → PostToolUse 钩子触发 → 注入内容传回模型并得到确认。

添加后端：在脚本的 BACKEND_CONFIGS 中添加一个条目：

"your_backend": {
    "type": "your_backend",
    "model": "your-model",
    "description": "Your Backend description",
    "api_style": "openai",  # or "anthropic", "gemini"
},

目前已覆盖：claude、openai、gemini（原生 SDK）、openrouter、grok 尚未覆盖：codex、claude_code、copilot、gemini_cli — 这些后端使用基于文件的 IPC 或 SDK 原生钩子，需要单独进行 E2E 钩子验证。

重要性：单元测试会验证钩子数据结构是否正确。此脚本会验证钩子是否在真实流式轮次中实际触发。后端可能通过所有单元测试，但在实时执行期间仍会静默丢弃钩子。

7.4 沙箱 / 路径权限测试（CLI/SDK 后端）⚠️ 必需

脚本：scripts/test_native_tools_sandbox.py

**每个具备内置文件/命令行工具的新 CLI/SDK 后端都必须进行此测试。**该脚本会让真实 Agent 在真实文件系统上运行，并使用唯一密钥验证权限边界是否真正生效，而不仅仅是强制执行代码是否存在。

# Test your new backend
uv run python scripts/test_native_tools_sandbox.py --backend your_backend

# Use LLM judge to detect subtle leakage
uv run python scripts/test_native_tools_sandbox.py --backend your_backend --llm-judge

该脚本会验证完整的权限矩阵：

区域	预期读取权限	预期写入权限
工作区（cwd）	✅ 允许	✅ 允许
可写上下文路径	✅ 允许	✅ 允许
只读上下文路径	✅ 允许	❌ 阻止
所有上下文之外	取决于后端	❌ 阻止
父目录	取决于后端	❌ 阻止
/tmp	✅ 允许	取决于后端

该脚本使用唯一密钥（写入各区域文件的 UUID）来检测未经授权的读取，即使操作“失败”也能检测 — 如果密钥字符串出现在模型的响应中，就意味着发生了泄漏，无论错误消息如何。

添加后端：在脚本的 BACKEND_CONFIGS 中添加一个条目：

"your_backend": {
    "module": "massgen.backend.your_backend",
    "class": "YourBackend",
    "model": "your-model",
    "blocks_reads_outside": True,   # Does your enforcement block reads?
    "blocks_tmp_writes": True,      # Does your enforcement block /tmp writes?
},

目前已覆盖：claude_code、codex 尚未覆盖：copilot、gemini_cli — 这些后端投入生产时必须加入测试。

重要性：权限回调和钩子单元测试会单独验证强制执行逻辑。此脚本会验证权限是否确实阻止实时 Agent 访问受限路径。后端可能具有正确的钩子逻辑，但如果钩子未接入正确的执行路径，仍可能允许未经授权的访问。

7.5 配置验证

uv run python scripts/validate_all_configs.py

阶段 8：配置示例（2 个以上文件）

8.1 单个 Agent

创建：massgen/configs/providers/<name>/single_<name>.yaml

8.2 多 Agent / 工具

创建：massgen/configs/providers/<name>/<name>_with_tools.yaml

阶段 9：文档（3 个以上文件）

9.1 YAML 架构

文件：docs/source/reference/yaml_schema.rst — 记录后端特定参数

9.2 后端表格

uv run python docs/scripts/generate_backend_tables.py

9.3 用户指南

文件：docs/source/user_guide/backends.rst — 添加新后端的章节

阶段 10：系统提示词注意事项

系统提示词由 massgen/system_message_builder.py 使用来自 massgen/system_prompt_sections.py 的优先级分区组装。不同分区会根据后端功能和配置标志有条件地加入。

10.1 章节类别

始终包含（所有后端）：

章节	用途
AgentIdentitySection	Agent 是谁
EvaluationSection	vote/new_answer 协调原语
CoreBehaviorsSection	行动倾向、并行执行
OutputFirstVerificationSection	质量迭代循环
SkillsSection	通过命令执行运行 openskills read
MemorySection	决策文档
WorkspaceStructureSection	工作区路径
ProjectInstructionsSection	CLAUDE.md/AGENTS.md
SubagentSection	MassGen 子 Agent 委派
BroadcastCommunicationSection	Agent 间通信
PostEvaluationSection	提交/重启
FileSearchSection	rg/sg 通用 CLI 工具
TaskContextSection	创建 CONTEXT.md
NoveltyPressureSection	各 Agent 间的新颖性/多样性压力
ChangedocSection	变更文档

由配置标志决定：

章节	条件
TaskPlanningSection	enable_task_planning=True
EvolvingSkillsSection	auto_discover_custom_tools=True 且 enable_task_planning=True
PlanningModeSection	planning_mode_enabled=True
CodeBasedToolsSection	enable_code_based_tools=True（CodeAct 范式）
CommandExecutionSection	enable_mcp_command_line=True
DecompositionSection	分解模式已启用
MultimodalToolsSection	enable_multimodal_tools=True

特定于模型：

章节	条件
GPT5GuidanceSection	仅限 GPT-5 模型
GrokGuidanceSection	仅限 Grok 模型

针对原生后端的调整：

章节	调整
FilesystemOperationsSection	has_native_tools=True → 使用通用工具措辞
FilesystemBestPracticesSection	调整比较工具的措辞

10.2 tool_category_overrides

文件：massgen/backend/native_tool_mixin.py

NativeToolBackendMixin 上的抽象方法 get_tool_category_overrides() 声明后端原生处理的 MassGen 工具类别。每个原生后端都会实现此方法：

@abstractmethod
def get_tool_category_overrides(self) -> Dict[str, str]:
    # "skip"     = backend has native equivalent
    # "override" = attach MassGen's version, disable native
    # (absent)   = attach MCP normally

类别：filesystem、command_execution、file_search、web_search、planning、subagents

目前由 system_message_builder.py 用于：

• 在 filesystem: "skip" 时，向文件系统章节传递 has_native_tools=True
• 记录每个后端原生处理的工具类别

注意：MCP 服务注入筛选（要附加哪些 MCP 服务）由各 CLI/SDK 后端（codex.py、claude_code.py）分别处理。get_tool_category_overrides() 方法用于文档说明，并控制系统提示词章节的行为。对于非 mixin 后端（基于 API 的后端），system_message_builder.py 的回退逻辑默认返回 {}。

10.3 路径权限集成

当 filesystem: "skip" 时，后端的原生工具会处理文件操作，但它们可能不会执行 MassGen 的 PathPermissionManager（PPM）精细路径权限。各后端对此采用不同的处理方式：

双层防护（建议用于支持权限回调的 SDK 后端）：

1. 第 1 层 — 权限回调（粗粒度门控）：从 SDK 的权限请求中提取路径，并根据 PPM 进行验证。如果无法提取路径，则放行（交由第 2 层处理）。
2. 第 2 层 — PreToolUse 钩子（细粒度控制）：对于非 Claude-Code 原生后端，编排器会自动将 PathPermissionManagerHook 注册为 PRE_TOOL_USE。此时可以获取完整的 toolName/toolArgs，以进行精确验证。

各后端的处理方式：

后端	PPM 策略
Claude Code	使用自身的 add_dirs 沙箱机制——编排器会跳过 PPM 钩子
Copilot	双层防护（权限回调 + PPM PreToolUse 钩子）。在 Docker 模式下，容器提供隔离，PPM 则作为纵深防御
Codex	不支持原生钩子——依赖 Docker/工作区隔离
API 后端	通过 ToolManager 自动执行 PPM

参考：双层模式可参见 copilot.py（_build_permission_callback）和 orchestrator.py（_setup_native_hooks_for_agent）。PPM 的内部实现见 massgen/filesystem_manager/_path_permission_manager.py。

阶段 11：钩子

MassGen 支持使用钩子拦截工具执行和其他生命周期事件。钩子会透明地运行（不会记录在系统提示词中）。

请注意，某些后端可能不支持钩子（例如 Codex）。在这种情况下，钩子只会在 MassGen 工具执行框架支持的范围内生效。

11.1 钩子类型与内置钩子

钩子类型（massgen/mcp_tools/hooks.py 中的 HookType 枚举）：

HookType	触发时机
PRE_TOOL_USE	工具执行前
POST_TOOL_USE	工具执行后

内置钩子类（PatternHook 子类——由编排器注册）：

Hook 类	类型	用途
PathPermissionManagerHook	PreToolUse	执行精细的路径读写权限（位于 _path_permission_manager.py）
RoundTimeoutPreHook	PreToolUse	执行轮次时间限制
HumanInputHook	PreToolUse	将人工输入注入 Agent 流程
MidStreamInjectionHook	PostToolUse	注入协调消息（投票提醒等）
SubagentCompleteHook	PostToolUse	子 Agent 完成时发出通知
BackgroundToolCompleteHook	PostToolUse	后台工具完成时发出通知
HighPriorityTaskReminderHook	PostToolUse	提醒 Agent 当前任务
MediaCallLedgerHook	PostToolUse	跟踪媒体生成调用
RoundTimeoutPostHook	PostToolUse	执行轮次时间限制

11.2 钩子架构

GeneralHookManager (massgen/mcp_tools/hooks/)
    ├── Registers global and per-agent hooks
    ├── Executes hooks in order for MCP-based backends
    └── For native backends → NativeHookAdapter
            ├── ClaudeCodeNativeHookAdapter
            │   └── Converts to Claude Agent SDK HookMatcher format
            └── (future) CodexNativeHookAdapter

11.3 原生钩子适配器——实现方法

对于使用原生工具执行的后端（CLI/SDK 包装器），需要将 MassGen 钩子桥接到后端的钩子系统中。这通过 NativeHookAdapter 实现（基类位于 massgen/mcp_tools/native_hook_adapters/base.py）。

第 1 步：创建适配器子类：

# massgen/mcp_tools/native_hook_adapters/your_backend_adapter.py
from .base import NativeHookAdapter

class YourBackendNativeHookAdapter(NativeHookAdapter):
    def supports_hook_type(self, hook_type):
        return hook_type in (HookType.PRE_TOOL_USE, HookType.POST_TOOL_USE)

    def convert_hook_to_native(self, hook, hook_type, context_factory=None):
        """Wrap MassGen hook as your backend's native hook format."""
        async def native_hook(tool_name, tool_args, context):
            event = self.create_hook_event_from_native(
                {"tool_name": tool_name, "tool_input": tool_args},
                hook_type, context_factory() if context_factory else {},
            )
            result = await hook.execute(event)
            return self.convert_hook_result_to_native(result, hook_type)
        return {"pattern": hook.matcher, "handler": native_hook}

    def build_native_hooks_config(self, hook_manager, agent_id=None, context_factory=None):
        """Convert all registered hooks to native config."""
        config = {"PreToolUse": [], "PostToolUse": []}
        for hook_type, hooks in hook_manager.get_hooks(agent_id).items():
            for hook in hooks:
                native = self.convert_hook_to_native(hook, hook_type, context_factory)
                config[hook_type.value].append(native)
        return config

    def merge_native_configs(self, *configs):
        """Merge permission hooks + MassGen hooks + user hooks."""
        merged = {"PreToolUse": [], "PostToolUse": []}
        for config in configs:
            for key in merged:
                merged[key].extend(config.get(key, []))
        return merged

第 2 步：在后端的 __init__ 中初始化：

from ..mcp_tools.native_hook_adapters import YourBackendNativeHookAdapter
self._native_hook_adapter = YourBackendNativeHookAdapter()

第 3 步：实现后端接口方法：

def supports_native_hooks(self) -> bool:
    return self._native_hook_adapter is not None

def get_native_hook_adapter(self):
    return self._native_hook_adapter

def set_native_hooks_config(self, config):
    self._massgen_hooks_config = config

第 4 步：启动后端进程时合并钩子。编排器会调用 set_native_hooks_config()，并传入转换后的钩子；你的后端在构建选项时，将这些钩子与自身的权限钩子合并。

11.4 参考

• Hook 框架：massgen/mcp_tools/hooks/
• 适配器基类：massgen/mcp_tools/native_hook_adapters/base.py
• Claude Code 适配器：massgen/mcp_tools/native_hook_adapters/claude_code_adapter.py
• Copilot 适配器：massgen/mcp_tools/native_hook_adapters/copilot_adapter.py
• 路径权限：massgen/filesystem_manager/_path_permission_manager.py
• Orchestrator Hook 设置：massgen/orchestrator.py（搜索 native_hook）

常见模式

CLI 包装器后端（如 Codex）

__init__: find CLI binary, set config defaults
stream_with_tools: build command -> spawn subprocess -> parse JSONL stdout -> yield StreamChunks
MCP: write project-scoped config file in workspace before launch
Custom tools: wrap as MCP server or inject via system prompt
Cleanup: remove workspace config on reset_state/clear_history

CLI 后端的 Docker 执行

如果基于 CLI 的后端不原生支持 Docker，请在 MassGen 容器内运行 CLI：

1. 在 Docker 镜像中安装 CLI（在 Dockerfile 的 npm install -g 中添加），或通过 command_line_docker_packages.preinstall.npm 安装
2. 在 __init__ 中检测 command_line_execution_mode: "docker"，并跳过主机 CLI 查找
3. 在 stream_with_tools 中，使用 container.client.api.exec_create() + exec_start(stream=True) 在容器内运行 CLI
4. 如果 CLI 自带沙箱，请将其禁用——容器已提供隔离
5. 如果 CLI 需要主机凭据，请通过 _build_credential_mounts() 以只读方式挂载

重要：某些 CLI 后端在 Docker 模式下运行时需要设置 command_line_docker_network_mode: bridge。这样容器才能向 API 发出出站网络请求。添加配置验证以强制执行此设置：

# In config_validator.py
if backend_type == "your_backend":
    execution_mode = backend_config.get("command_line_execution_mode")
    if execution_mode == "docker":
        if "command_line_docker_network_mode" not in backend_config:
            result.add_error(
                "YourBackend in Docker mode requires 'command_line_docker_network_mode'",
                f"{location}.command_line_docker_network_mode",
                "Add 'command_line_docker_network_mode: bridge' (required for network access)",
            )

参考：codex.py 通过 _stream_docker() / _stream_local() 分支实现此模式。

SDK 后端的 Docker 执行（如 Copilot）

基于 SDK 的后端（进程内 Python 客户端）无法在 Docker 内运行——SDK 保留在主机上。相反，Docker 通过以下方式隔离文件/终端操作：

1. 在 Docker 模式下通过 excluded_tools 禁用内置工具，以免 SDK 使用其原生文件/终端工具
2. 将所有文件/终端操作路由到 MCP 服务，由其在 Docker 容器内执行
3. 将 working_directory 指向 Docker 挂载的工作区路径

# In __init__:
self._docker_execution = (
    kwargs.get("command_line_execution_mode") == "docker"
    or self.config.get("command_line_execution_mode") == "docker"
)

# Property to check if Docker is actually active:
@property
def _is_docker_mode(self) -> bool:
    if not self._docker_execution:
        return False
    if not self.filesystem_manager:
        return False
    dm = getattr(self.filesystem_manager, "docker_manager", None)
    if dm is None:
        return False
    agent_id = self.config.get("agent_id")
    if agent_id and dm.get_container(agent_id):
        return True
    return False

# In get_disallowed_tools():
def get_disallowed_tools(self, config):
    if self._docker_execution:
        return ["editFile", "createFile", "deleteFile",
                "readFile", "listDirectory",
                "runShellCommand", "shellCommand"]
    return []

# In stream_with_tools(), merge Docker-excluded tools into session config:
if self._docker_execution:
    docker_excluded = self.get_disallowed_tools(self.config)
    if docker_excluded:
        excluded_tools = list(set((excluded_tools or []) + docker_excluded))

与 CLI Docker 模式的主要区别：

• SDK 保留在主机上（不使用 exec_create / exec_start）
• 通过 SDK 会话配置（excluded_tools）禁用内置工具，而不是在容器内运行
• MCP 服务仍在容器内执行（与 CLI 模式相同）
• PPM 权限回调仍处于启用状态，作为纵深防御

配置验证：采用相同模式——要求设置 command_line_docker_network_mode：

if backend_type == "copilot":
    execution_mode = backend_config.get("command_line_execution_mode")
    if execution_mode == "docker":
        if "command_line_docker_network_mode" not in backend_config:
            result.add_error(...)

参考：copilot.py 实现了此模式。

Docker MCP 路径解析

MCP 服务配置由主机上的 orchestrator 使用绝对主机文件路径构建（例如，fastmcp run /host/path/massgen/mcp_tools/planning/_server.py:create_server）。在 Docker 内部，除非显式挂载，否则这些路径不存在。

解决方案：_docker_manager.py 会将 massgen/ 包目录以只读方式绑定挂载到容器内的相同主机路径。这样，基于主机路径的 MCP 配置可在容器内直接使用，并采用主机上的最新源代码，而不是过时的 pip 安装模块。

基于原生工具的后端 Shell 沙盒

本节仅适用于自带 Shell/文件工具的后端（例如内置命令执行功能的 CLI 封装）。如果后端将所有工具执行都委托给 MassGen 的 MCP 服务，则 MassGen 的 PathPermissionManager 会自动处理沙盒配置。

当后端拥有可访问文件系统或运行命令的自有工具时，必须将 MassGen 的权限模型映射到后端的沙盒机制：

MassGen 概念	预期访问权限
工作区	写入
临时工作区	只读
读取上下文路径	只读
写入上下文路径	写入 — 必须在后端沙盒配置中明确授权

通常情况下，工作区默认可写。一般也允许读取整个文件系统。关键任务是确保将写入上下文路径添加到后端的可写允许列表中（例如 Codex 的 sandbox_workspace_write.writable_roots，或 Claude Code 的 allowed_directories）。

Docker 模式：跳过后端沙盒配置 — 容器本身就是沙盒。在容器内授予完全访问权限。

SDK 封装后端（如 Claude Code、Copilot）

__init__: import SDK, configure options, detect Docker mode
stream_with_tools: call SDK with messages -> iterate events -> yield StreamChunks
MCP: pass mcp_servers dict to SDK options
Custom tools: create SDK MCP server from tool definitions
State: SDK manages conversation state internally
Docker: SDK stays on host; disable built-in file/shell tools via excluded_tools;
        route file/shell ops through MCP servers in container (see §Docker Execution for SDK Backends)

API 后端（如 Claude、Gemini）

Subclass CustomToolAndMCPBackend or ChatCompletionsBackend
Implement formatter if non-standard API format
Implement API params handler to filter/transform config -> API params
MCP + custom tools handled automatically by base class

按复杂度划分的参考后端

后端	行数	类型	适合参考的内容
grok.py	~81	ChatCompletions 子类	最简 API 后端
chat_completions.py	~1150	OpenAI 兼容	标准 API 后端
response.py	~1600	Response API	推理模型
claude.py	~1830	自定义 API	功能齐全的 API
copilot.py	~1250	SDK 封装	基于 SDK + PPM 权限回调 + Docker 模式
codex.py	~2090	CLI 封装	基于 CLI + Docker
gemini.py	~2460	自定义 API	非 OpenAI API
claude_code.py	~3530	SDK 封装	功能齐全的有状态后端

文件摘要

#	文件	必需？	用途
1	massgen/backend/<name>.py	是	核心后端
2	massgen/formatter/<name>_formatter.py	如果使用非标准 API	消息/工具格式化
3	massgen/api_params_handler/<name>_handler.py	如果使用 HTTP API	参数筛选
4	massgen/backend/__init__.py	是	导出
5	massgen/cli.py	是	后端创建
6	massgen/backend/capabilities.py	是	模型注册表
7	massgen/config_validator.py	如果有特殊规则	验证
8	massgen/token_manager/token_manager.py	如果 LiteLLM 中没有	定价
9	massgen/backend/base.py	如果有新的 YAML 参数	排除参数
10	massgen/api_params_handler/_base.py	如果有新的 YAML 参数	排除参数
11	massgen/tests/test_*	是	测试
12	massgen/configs/providers/<name>/	是	示例配置
13	docs/source/reference/yaml_schema.rst	是	Schema 文档
14	docs/source/user_guide/backends.rst	是	用户指南
15	docs/scripts/generate_backend_tables.py	运行该脚本	重新生成表格