backend-integrator

What this skill does

# 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**:
```python
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:
```python
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:
```python
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:
```python
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:
```python
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`

```python
from .your_backend import YourBackend
# Add to __all__
```

#### 2.2 CLI Backend Mapping
**File**: `massgen/cli.py`

Add to `create_backend()` function:
```python
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`:
```python
"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 Config Validator (if needed)
**File**: `massgen/config_validator.py`

Add backend-specific validation to `_validate_backend()` if there are special rules (e.g., required params, incompatible combinations).

### Phase 3: Token Management (1 file)

#### 3.1 Pricing
**File**: `massgen/token_manager/token_manager.py`

**Check LiteLLM first** — only add to `PROVIDER_PRICING` if the model is NOT in the LiteLLM database. Provider name must match `get_provider_name()` exactly (case-sensitive).

### Phase 4: Excluded Params (2 files, if adding new YAML params)

#### 4.1 Base Class Exclusions
**File**: `massgen/backend/base.py` -> `get_base_excluded_config_params()`

#### 4.2 API Params Handler Exclusions
**File**: `massgen/api_params_handler/_api_params_handler_base.py` -> `get_base_excluded_params()`

Both must stay in sync. Add any new framework-level YAML params that should NOT be passed to the provider API.

### Phase 5: Authentication

#### 5.1 API Key Backends (standard)
Most backends use API keys. Set `env_var` in `capabilities.py` and add the key check in `cli.py`:
```python
# 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 / Subscription Auth (CLI/SDK wrappers)
For backends that support OAuth (like Codex, Claude Code), implement a multi-tier auth cascade:

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