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

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."""

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

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

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

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

from .your_backend import YourBackend

undefined

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)

"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
)

undefined

undefined

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, ...)

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

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

from .base import get_multimodal_tool_definitions

- **API backends** (`CustomToolAndMCPBackend` subclasses): handled automatically — base class calls `get_multimodal_tool_definitions()` and registers via `_register_custom_tools()`
- **CLI/SDK backends** (`LLMBackend` subclasses like Codex, Claude Code): must do this explicitly in `__init__`, then wrap as MCP server (see §6.4)

**Important**: Always use `get_multimodal_tool_definitions()` — never inline the tool dicts. This keeps the definitions in one place.

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

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

undefined

**For CLI wrappers without SDK MCP** (like Codex):
Use the shared `massgen/mcp_tools/custom_tools_server.py` utility. It creates a standalone fastmcp server that wraps ToolManager tools via stdio transport.

```python
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)

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

from .base import build_workflow_instructions, parse_workflow_tool_calls, extract_structured_response

**For API backends**: Workflow tools are passed as normal function tools — no injection needed.

**For CLI/SDK backends** (Codex, Claude Code): The model can't receive native function tool definitions, so:

1. **Build instructions**: Call `build_workflow_instructions(tools)` to get the instruction text
2. **Inject into system prompt**: Append to whatever mechanism the backend uses for system prompts
3. **Accumulate text output**: Track all `content` chunks during streaming
4. **Parse after streaming**: Call `parse_workflow_tool_calls(accumulated_text)` to extract tool calls
5. **Yield as done chunk**: `yield StreamChunk(type="done", tool_calls=workflow_tool_calls)`

**Codex-specific**: Instructions go into `AGENTS.md` at the workspace root (Codex auto-reads this). The orchestrator's system message is also extracted from messages and included, since Codex only receives a single user prompt via CLI — the system message from the orchestrator would otherwise be lost. This approach was utilized as the developer instructions approach wasn't working.

**Claude Code-specific**: Instructions are built by `_build_system_prompt_with_workflow_tools()` which also adds tool calling sections, then passed as `system_prompt` to the SDK.

It is an open question as to whether typical workflow tool parsing works (e.g., Claude Code) or if we need to use MCP tools (e.g., Codex).

undefined

**Key patterns**:
- Claude Code SDK supports glob patterns for Bash restrictions: `"Bash(rm*)"`, `"Bash(sudo*)"`
- Check `if "disallowed_tools" not in all_params` to allow user override via YAML config
- Codex uses `disabled_tools` per MCP server in `.codex/config.toml`; for built-in tools, Codex doesn't have a disable mechanism — use `--full-auto` to auto-approve instead

**Document in capabilities.py** which native tools the backend provides:
```python
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."

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"]

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

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"},
}

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

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

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

undefined

The script verifies the full permission matrix:

| Zone | Expected reads | Expected writes |
|------|---------------|-----------------|
| Workspace (cwd) | ✅ allowed | ✅ allowed |
| Writable context path | ✅ allowed | ✅ allowed |
| Read-only context path | ✅ allowed | ❌ blocked |
| Outside all contexts | depends on backend | ❌ blocked |
| Parent directory | depends on backend | ❌ blocked |
| `/tmp` | ✅ allowed | depends on backend |

It uses **unique secrets** (UUIDs written to each zone's files) to detect unauthorized reads even when the operation "fails" — if the secret string appears in the model's response, there's a leak regardless of error messages.

**Adding your backend**: Add an entry to `BACKEND_CONFIGS` in the script:
```python
"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?
},

uv run python scripts/validate_all_configs.py

uv run python docs/scripts/generate_backend_tables.py

@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

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

undefined

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

**Step 2**: Initialize in backend `__init__`:
```python
from ..mcp_tools.native_hook_adapters import YourBackendNativeHookAdapter
self._native_hook_adapter = YourBackendNativeHookAdapter()

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

__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

undefined

**Reference**: `codex.py` implements this pattern with `_stream_docker()` / `_stream_local()` branching.

undefined

**Key differences from CLI Docker pattern**:
- SDK stays on host (no `exec_create` / `exec_start`)
- Built-in tools disabled via SDK session config (`excluded_tools`), not by running inside container
- MCP servers still execute inside container (same as CLI pattern)
- PPM permission callback still active as defense-in-depth

**Config validation**: Same pattern — require `command_line_docker_network_mode`:
```python
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(...)

__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)

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