/backend-integrator

name: backend-integrator
description: Complete guide for integrating a new LLM backend into MassGen. Use when adding a new provider (e.g., Codex, Mistral, DeepSeek) or when auditing an existing backend for missing integration points. Covers all ~15 files that need touching.

Backend Integrator

This skill provides the complete checklist and patterns for integrating a new LLM backend into MassGen. A full integration touches ~15 files across the codebase.

When to Use This Skill

• Adding a new LLM provider/backend
• Auditing an existing backend for missing integration points
• Understanding what files to modify when extending backend capabilities

Integration Architecture

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

Complete Checklist

Phase 1: Core Implementation (3 files)

1.1 Backend Class

**File**: `massgen/backend/<name>.py`

**Choose base class**:

• `LLMBackend` — bare minimum, you handle everything
• `CustomToolAndMCPBackend` — adds MCP + custom tool support (most common)
• `ChatCompletionsBackend` — for OpenAI-compatible APIs (inherits from above)
• `ResponseBackend` — for OpenAI Response API format

**Required methods**:

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 types to yield**: | Type | When | Key fields | |------|------|------------| | `"content"` | Text output | `content="..."` | | `"tool_calls"` | Tool invocation | `tool_calls=[{id, name, arguments}]` | | `"reasoning"` | Thinking/reasoning delta | `reasoning_delta="..."` | | `"reasoning_done"` | Reasoning complete | `reasoning_text="..."` | | `"reasoning_summary"` | Reasoning summary delta | `reasoning_summary_delta="..."` | | `"reasoning_summary_done"` | Reasoning summary complete | `reasoning_summary_text="..."` | | `"complete_message"` | Full assistant message | `complete_message={...}` | | `"complete_response"` | Raw API response | `response={...}` | | `"done"` | Stream complete | `usage={prompt_tokens, completion_tokens, total_tokens}` | | `"error"` | Error occurred | `error="..."` | | `"agent_status"` | Status update | `status="...", detail="..."` | | `"backend_status"` | Backend-level status | `status="...", detail="..."` | | `"compression_status"` | Compression event | `status="...", detail="..."` | | `"hook_execution"` | Hook ran | `hook_info={...}, tool_call_id="..."` |

**Common fields on all chunks**: `source` (agent/orchestrator ID), `display` (bool, default True).

**Token tracking** — call one of:

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

**Timing** — call in 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

**For stateful backends** (CLI/SDK wrappers), also implement:

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

**Compression support** — inherit `StreamingBufferMixin` and call:

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

1.2 Formatter (if needed)

**File**: `massgen/formatter/<name>_formatter.py`

Only needed if the API uses a non-standard message/tool format (not OpenAI chat completions format). Subclass `FormatterBase` and implement `format_messages()`, `format_tools()`, `format_mcp_tools()`.

Existing formatters:

• `_claude_formatter.py` — Anthropic Messages API
• `_gemini_formatter.py` — Gemini API
• `_chat_completions_formatter.py` — OpenAI/generic (reuse for compatible APIs)
• `_response_formatter.py` — OpenAI Response API format

1.3 API Params Handler (if needed)

**File**: `massgen/api_params_handler/<name>_api_params_handler.py`

Only needed if the backend calls an HTTP API and needs to filter/transform YAML config params before passing to the API. Subclass `APIParamsHandlerBase`.

CLI/SDK wrappers (Codex, Claude Code) typically don't need this — they build commands directly.

Phase 2: Registration (4 files)

2.1 Backend __init__.py

**File**: `massgen/backend/__init__.py`

from .your_backend import YourBackend
# Add to __all__

2.2 CLI Backend Mapping

**File**: `massgen/cli.py`

Add to `create_backend()` function:

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)

For CLI-based backends that don't need API keys, skip the key check.

2.3 Capabilities Registry

**File**: `massgen/backend/capabilities.py`

Add entry to `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="mode