---

name: adapter-factory description: Guide for creating new CLI or HTTP adapters to integrate AI models into the AI Counsel deliberation system

Adapter Factory Skill

This skill teaches how to integrate new AI models into the AI Counsel MCP server by creating adapters. There are two types of adapters:

1. CLI Adapters - For command-line AI tools (e.g., claude, droid, codex)
2. HTTP Adapters - For HTTP API integrations (e.g., Ollama, LM Studio, OpenRouter)

When to Use This Skill

Use this skill when you need to:

• Add support for a new AI model or service to participate in deliberations
• Integrate a command-line AI tool (CLI adapter)
• Integrate an HTTP-based AI API (HTTP adapter)
• Understand the adapter pattern used in AI Counsel
• Debug or modify existing adapters

Architecture Overview

Base Classes

BaseCLIAdapter (adapters/base.py)

• Handles subprocess execution, timeout management, and error handling
• Provides invoke() method that manages the full CLI lifecycle
• Subclasses only implement parse_output() for tool-specific parsing
• Optional: Override _adjust_args_for_context() for deliberation vs. non-deliberation behavior
• Optional: Implement validate_prompt_length() for length limits

BaseHTTPAdapter (adapters/base_http.py)

• Handles HTTP requests, retry logic with exponential backoff, and timeout management
• Uses httpx for async HTTP and tenacity for retry logic
• Retries on 5xx/429/network errors, fails fast on 4xx client errors
• Subclasses implement build_request() and parse_response()
• Supports environment variable substitution for API keys

Factory Pattern

The create_adapter() function in adapters/__init__.py:

• Maintains registries for CLI and HTTP adapters
• Creates appropriate adapter instances from config
• Handles backward compatibility with legacy config formats

---

Creating a CLI Adapter

Follow these 6 steps to add a new command-line AI tool:

Step 1: Create Adapter File

Create adapters/your_cli.py:

"""Your CLI tool adapter."""
from adapters.base import BaseCLIAdapter


class YourCLIAdapter(BaseCLIAdapter):
    """Adapter for your-cli tool."""

    def parse_output(self, raw_output: str) -> str:
        """
        Parse your-cli output to extract model response.

        Args:
            raw_output: Raw stdout from CLI tool

        Returns:
            Parsed model response text
        """
        # Example: Strip headers and extract main response
        lines = raw_output.strip().split("\n")

        # Skip header lines (tool-specific logic)
        start_idx = 0
        for i, line in enumerate(lines):
            if line.strip() and not line.startswith("Loading"):
                start_idx = i
                break

        return "\n".join(lines[start_idx:]).strip()

Optional: Override _adjust_args_for_context() if your CLI needs different args for deliberation vs. regular use:

def _adjust_args_for_context(self, is_deliberation: bool) -> list[str]:
    """
    Adjust CLI arguments based on context.

    Args:
        is_deliberation: True if part of multi-model deliberation

    Returns:
        Adjusted argument list
    """
    args = self.args.copy()

    if is_deliberation:
        # Remove flags that interfere with deliberation
        if "--interactive" in args:
            args.remove("--interactive")
    else:
        # Add flags for regular Claude Code work
        if "--context" not in args:
            args.append("--context")

    return args

Optional: Implement validate_prompt_length() if your API has length limits:

MAX_PROMPT_CHARS = 100000  # Example: 100k char limit

def validate_prompt_length(self, prompt: str) -> bool:
    """
    Validate prompt length against API limits.

    Args:
        prompt: The full prompt to validate

    Returns:
        True if valid, False if too long
    """
    return len(prompt) <= self.MAX_PROMPT_CHARS

Step 2: Update Config

Add your CLI to config.yaml:

adapters:
  your_cli:
    type: cli
    command: "your-cli"
    args: ["--model", "{model}", "{prompt}"]
    timeout: 60  # Adjust based on your model's speed

Placeholder Variables:

• {model} - Replaced with model identifier
• {prompt} - Replaced with full prompt text (including context)

Timeout Guidelines:

• Fast models (GPT-3.5): 30-60s
• Reasoning models (Claude Sonnet 4.5, GPT-5): 180-300s
• Local models: Varies, test and adjust

Step 3: Register Adapter

Update adapters/__init__.py:

# Add import at top
from adapters.your_cli import YourCLIAdapter

# Add to cli_adapters dict in create_adapter()
cli_adapters: dict[str, Type[BaseCLIAdapter]] = {
    "claude": ClaudeAdapter,
    "codex": CodexAdapter,
    "droid": DroidAdapter,
    "gemini": GeminiAdapter,
    "llamacpp": LlamaCppAdapter,
    "your_cli": YourCLIAdapter,  # Add this line
}

# Add to __all__ export list
__all__ = [
    "BaseCLIAdapter",
    "BaseHTTPAdapter",
    "ClaudeAdapter",
    "CodexAdapter",
    "DroidAdapter",
    "GeminiAdapter",
    "LlamaCppAdapter",
    "YourCLIAdapter",  # Add this line
    # ... rest of exports
]

Step 4: Update Schema

Update models/schema.py:

# Find the Participant class and add your CLI to the Literal type
class Participant(BaseModel):
    cli: Literal[
        "claude",
        "codex",
        "droid",
        "gemini",
        "llamacpp",
        "your_cli"  # Add this
    ]
    model: str

Update the MCP tool description in server.py:

# Find RECOMMENDED_MODELS and add your models
RECOMMENDED_MODELS = {
    "claude": ["sonnet-4.5", "opus-4"],
    "codex": ["gpt-5-codex", "gpt-4o"],
    # ... other models
    "your_cli": ["your-model-1", "your-model-2"],  # Add this
}

Step 5: Add Recommended Models

Update server.py::RECOMMENDED_MODELS with suggested models for your CLI:

RECOMMENDED_MODELS = {
    # ... existing models
    "your_cli": [
        "your-fast-model",      # For quick responses
        "your-reasoning-model",  # For complex analysis
    ],
}

Step 6: Write Tests

Create unit tests in tests/unit/test_adapters.py:

import pytest
from adapters.your_cli import YourCLIAdapter


class TestYourCLIAdapter:
    def test_parse_output_basic(self):
        """Test basic output parsing."""
        adapter = YourCLIAdapter(
            command="your-cli",
            args=["--model", "{model}", "{prompt}"],
            timeout=60
        )

        raw_output = "Loading...\n\nActual response text here"
        result = adapter.parse_output(raw_output)

        assert result == "Actual response text here"
        assert "Loading" not in result

    def test_parse_output_multiline(self):
        """Test multiline response parsing."""
        adapter = YourCLIAdapter(
            command="your-cli",
            args=["--model", "{model}", "{prompt}"],
            timeout=60
        )

        raw_output = "Header\n\nLine 1\nLine 2\nLine 3"
        result = adapter.parse_output(raw_output)

        assert "Line 1" in result
        assert "Line 2" in result
        assert "Line 3" in result

Add integration tests in tests/integration/:

import pytest
from adapters.your_cli import YourCLIAdapter


@pytest.mark.integration
@pytest.mark.asyncio
async def test_your_cli_integration():
    """Test actual CLI invocation (requires your-cli installed)."""
    adapter = YourCLIAdapter(
        command="your-cli",
        args=["--model", "{model}", "{prompt}"],
        timeout=60
    )

    result = await adapter.invoke(
        prompt="What is 2+2?",
        model="your-default-model"
    )

    assert result
    assert len(result) > 0
    # Add assertions specific to your model's response format

---

Creating an HTTP Adapter

Follow these 6 steps to add a new HTTP API integration:

Step 1: Create Adapter File

Create adapters/your_adapter.py:

"""Your API adapter."""
from typing import Tuple
from adapters.base_http import BaseHTTPAdapter


class YourAdapter(BaseHTTPAdapter):
    """
    Adapter for Your AI API.

    API reference: https://docs.yourapi.com
    Default endpoint: https://api.yourservice.com

    Example:
        adapter = YourAdapter(
            base_url="https://api.yourservice.com",
            api_key="your-key",
            timeout=120
        )
        result = await adapter.invoke(prompt="Hello", model="your-model")
    """

    def build_request(
        self, model: str, prompt: str
    ) -> Tuple[str, dict[str, str], dict]:
        """
        Build API request components.

        Args:
            model: Model identifier (e.g., "your-model-v1")
            prompt: The prompt to send

        Returns:
            Tuple of (endpoint, headers, body):
            - endpoint: URL path (e.g., "/v1/chat/completions")
            - headers: Request headers dict
            - body: Request body dict (will be JSON-encoded)
        """
        endpoint = "/v1/chat/completions"

        headers = {
            "Content-Type": "application/json",
        }

        # Add authentication if API key provided
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"

        # Build request body (adapt to your API format)
        body = {
            "model": model,
            "messages": [
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.7,
            "stream": False,
        }

        return (endpoint, headers, body)

    def parse_response(self, response_json: dict) -> str:
        """
        Parse API response to extract model output.

        Your API response format:
        {
          "id": "resp-123",
          "model": "your-model",
          "choices": [
            {
              "message": {
                "role": "assistant",
                "content": "The model's response text"
              }
            }
          ]
        }

        Args:
            response_json: Parsed JSON response from API

        Returns:
            Extracted model response text

        Raises:
            KeyError: If response format is unexpected
        """
        try:
            return response_json["choices"][0]["message"]["content"]
        except (KeyError, IndexError) as e:
            raise KeyError(
                f"Unexpected API response format. "
                f"Expected 'choices[0].message.content', "
                f"got keys: {list(response_json.keys())}"
            ) from e

Step 2: Update Config

Add your HTTP adapter to config.yaml:

adapters:
  your_adapter:
    type: http
    base_url: "https://api.yourservice.com"
    api_key: "${YOUR_API_KEY}"  # Environment variable substitution
    timeout: 120
    max_retries: 3

Configuration Options:

• base_url: Base URL for API (no trailing slash)
• api_key: API key (use ${ENV_VAR} for environment variables)
• timeout: Request timeout in seconds
• max_retries: Max retry attempts for 5xx/429/network errors (default: 3)
• headers: Optional default headers dict

Environment Variable Substitution:

• Pattern: ${VAR_NAME} in config
• Substituted at runtime from environment
• Secure: Keeps secrets out of config files

Step 3: Register Adapter

Update adapters/__init__.py:

# Add import at top
from adapters.your_adapter import YourAdapter

# Add to http_adapters dict in create_adapter()
http_adapters: dict[str, Type[BaseHTTPAdapter]] = {
    "ollama": OllamaAdapter,
    "lmstudio": LMStudioAdapter,
    "openrouter": OpenRouterAdapter,
    "your_adapter": YourAdapter,  # Add this line
}

# Add to __all__ export list
__all__ = [
    "BaseCLIAdapter",
    "BaseHTTPAdapter",
    # ... CLI adapters
    "OllamaAdapter",
    "LMStudioAdapter",
    "OpenRouterAdapter",
    "YourAdapter",  # Add this line
    "create_adapter",
]

Step 4: Set Environment Variables

If your adapter uses API keys:

# Add to your shell profile (~/.bashrc, ~/.zshrc, etc.)
export YOUR_API_KEY="your-actual-api-key-here"

# Or set temporarily for testing
export YOUR_API_KEY="test-key" && python server.py

Step 5: Write Tests

Create unit tests with VCR for HTTP response recording in tests/unit/test_your_adapter.py:

import pytest
import vcr
from adapters.your_adapter import YourAdapter


# Configure VCR for recording HTTP interactions
vcr_instance = vcr.VCR(
    cassette_library_dir="tests/fixtures/vcr_cassettes/your_adapter/",
    record_mode="once",  # Record once, then replay
    match_on=["method", "scheme", "host", "port", "path", "query"],
    filter_headers=["authorization"],  # Hide API keys in recordings
)


class TestYourAdapter:
    def test_build_request_basic(self):
        """Test request building without API key."""
        adapter = YourAdapter(
            base_url="https://api.example.com",
            timeout=60
        )

        endpoint, headers, body = adapter.build_request(
            model="test-model",
            prompt="Hello"
        )

        assert endpoint == "/v1/chat/completions"
        assert headers["Content-Type"] == "application/json"
        assert body["model"] == "test-model"
        assert body["messages"][0]["content"] == "Hello"

    def test_build_request_with_auth(self):
        """Test request building with API key."""
        adapter = YourAdapter(
            base_url="https://api.example.com",
            api_key="test-key",
            timeout=60
        )

        endpoint, headers, body = adapter.build_request(
            model="test-model",
            prompt="Hello"
        )

        assert "Authorization" in headers
        assert headers["Authorization"] == "Bearer test-key"

    def test_parse_response_success(self):
        """Test parsing successful response."""
        adapter = YourAdapter(
            base_url="https://api.example.com",
            timeout=60
        )

        response = {
            "id": "resp-123",
            "choices": [
                {
                    "message": {
                        "role": "assistant",
                        "content": "Hello! How can I help?"
                    }
                }
            ]
        }

        result = adapter.parse_response(response)
        assert result == "Hello! How can I help?"

    def test_parse_response_missing_field(self):
        """Test error handling for malformed response."""
        adapter = YourAdapter(
            base_url="https://api.example.com",
            timeout=60
        )

        response = {"id": "resp-123"}  # Missing choices

        with pytest.raises(KeyError) as exc_info:
            adapter.parse_response(response)

        assert "Unexpected API response format" in str(exc_info.value)

    @pytest.mark.asyncio
    @vcr_instance.use_cassette("invoke_basic.yaml")
    async def test_invoke_with_vcr(self):
        """Test full invoke() with VCR recording."""
        adapter = YourAdapter(
            base_url="https://api.example.com",
            api_key="test-key",
            timeout=60
        )

        result = await adapter.invoke(
            prompt="What is 2+2?",
            model="test-model"
        )

        assert result
        assert len(result) > 0

Optional: Add integration tests (requires running service):

@pytest.mark.integration
@pytest.mark.asyncio
async def test_your_adapter_live():
    """Test with live API (requires service running and API key)."""
    import os

    api_key = os.getenv("YOUR_API_KEY")
    if not api_key:
        pytest.skip("YOUR_API_KEY not set")

    adapter = YourAdapter(
        base_url="https://api.yourservice.com",
        api_key=api_key,
        timeout=120
    )

    result = await adapter.invoke(
        prompt="What is the capital of France?",
        model="your-model"
    )

    assert "Paris" in result or "paris" in result

Step 6: Test with Deliberation

Create a simple test script to verify integration:

"""Test your adapter in a deliberation context."""
import asyncio
from adapters.your_adapter import YourAdapter


async def test():
    adapter = YourAdapter(
        base_url="https://api.yourservice.com",
        api_key="your-key",
        timeout=120
    )

    # Test basic invocation
    result = await adapter.invoke(
        prompt="What is 2+2?",
        model="your-model"
    )
    print(f"Response: {result}")

    # Test with context (simulating Round 2+ in deliberation)
    result_with_context = await adapter.invoke(
        prompt="Do you agree with this answer?",
        model="your-model",
        context="Previous participant said: 2+2 equals 4"
    )
    print(f"Response with context: {result_with_context}")


if __name__ == "__main__":
    asyncio.run(test())

---

Key Design Principles

DRY (Don't Repeat Yourself)

• Common logic in base classes (BaseCLIAdapter, BaseHTTPAdapter)
• Tool-specific logic in concrete adapters
• Only implement what's unique to your adapter

YAGNI (You Aren't Gonna Need It)

• Build only what's needed for basic integration
• Don't add features until they're required
• Start simple, extend as needed

TDD (Test-Driven Development)

• Write tests first (red)
• Implement adapter (green)
• Refactor for clarity (refactor)

Type Safety

• Use type hints throughout
• Pydantic validation where applicable
• Let mypy catch errors early

Error Isolation

• Adapter failures don't halt deliberations
• Other participants continue if one fails
• Graceful degradation with informative errors

---

Common Patterns

CLI Adapter with Custom Parsing

def parse_output(self, raw_output: str) -> str:
    """Parse CLI output with multiple header formats."""
    lines = raw_output.strip().split("\n")

    # Skip all header lines until we find content
    content_started = False
    result_lines = []

    for line in lines:
        # Detect header patterns
        if any(marker in line.lower() for marker in ["loading", "initializing", "version"]):
            continue

        # Detect content start
        if line.strip():
            content_started = True

        if content_started:
            result_lines.append(line)

    return "\n".join(result_lines).strip()

HTTP Adapter with Streaming Support

def build_request(self, model: str, prompt: str) -> Tuple[str, dict, dict]:
    """Build request with optional streaming."""
    # Note: BaseHTTPAdapter doesn't support streaming yet
    # This is for future extension

    body = {
        "model": model,
        "prompt": prompt,
        "stream": False,  # Keep False for now
    }

    return ("/api/generate", {"Content-Type": "application/json"}, body)

Environment Variable Validation

def __init__(self, base_url: str, api_key: str = None, **kwargs):
    """Initialize with API key validation."""
    if not api_key:
        raise ValueError(
            "API key required. Set YOUR_API_KEY environment variable or "
            "provide api_key parameter."
        )

    super().__init__(base_url=base_url, api_key=api_key, **kwargs)

---

Testing Guidelines

Unit Tests (Required)

• Test parse_output() with various input formats
• Test build_request() with different parameters
• Test parse_response() with success and error cases
• Mock external dependencies (no actual API calls)
• Fast execution (< 1s total)

Integration Tests (Optional)

• Test actual CLI/API invocation
• Requires tool installed or service running
• Mark with @pytest.mark.integration
• May be slow, use sparingly

VCR for HTTP Tests

• Record real HTTP interactions once
• Replay from cassettes in CI/CD
• Filter sensitive data (API keys, auth tokens)
• Store in tests/fixtures/vcr_cassettes/your_adapter/

E2E Tests (Optional)

• Full deliberation with your adapter
• Mark with @pytest.mark.e2e
• Very slow, expensive (real API calls)
• Use for final validation only

---

Troubleshooting

CLI Adapter Issues

Problem: Timeout errors

• Solution: Increase timeout in config.yaml
• Reasoning models need 180-300s, not 60s

Problem: Output parsing fails

• Solution: Print raw_output and examine format
• Each CLI has unique output format, adjust parsing logic

Problem: Hook interference (Claude CLI)

• Solution: Add --settings '{"disableAllHooks": true}' to args
• User hooks can interfere with deliberation invocations

HTTP Adapter Issues

Problem: Connection refused

• Solution: Verify base_url and service is running
• Check with curl $BASE_URL/health or similar

Problem: 401 Authentication errors

• Solution: Verify API key is set: echo $YOUR_API_KEY
• Check environment variable substitution in config

Problem: 400 Bad Request errors

• Solution: Log request body, check API documentation
• Common issue: Wrong field names or missing required fields

Problem: Retries exhausted

• Solution: Check if service is healthy
• 5xx errors trigger retries, 4xx do not (by design)

General Issues

Problem: Adapter not found

• Solution: Verify registration in adapters/__init__.py
• Check both import and dict addition

Problem: Schema validation errors

• Solution: Add CLI name to Participant.cli Literal in models/schema.py
• MCP won't accept unlisted CLI names

---

Reference Files

Essential Files

• adapters/base.py - CLI adapter base class
• adapters/base_http.py - HTTP adapter base class
• adapters/__init__.py - Adapter factory and registry
• models/config.py - Configuration schema
• models/schema.py - Data models and validation

Example Adapters

• adapters/claude.py - CLI adapter with context-aware args
• adapters/gemini.py - CLI adapter with length validation
• adapters/ollama.py - HTTP adapter for local API
• adapters/lmstudio.py - HTTP adapter with OpenAI-compatible format

Test Examples

• tests/unit/test_adapters.py - CLI adapter unit tests
• tests/unit/test_ollama.py - HTTP adapter unit tests with VCR
• tests/integration/test_cli_adapters.py - CLI integration tests

---

Next Steps After Creating Adapter

1. Update CLAUDE.md if you added new patterns or gotchas
2. Add to RECOMMENDED_MODELS in server.py with usage guidance
3. Document API quirks in adapter docstrings for future maintainers
4. Test in real deliberation with 2-3 participants
5. Monitor transcript for response quality and voting behavior
6. Share findings if you discover best practices for your model

---

Quick Reference

CLI Adapter Checklist

• Create adapters/your_cli.py with parse_output()
• Add to config.yaml with command, args, timeout
• Register in adapters/__init__.py (import + dict + export)
• Add to Participant.cli Literal in models/schema.py
• Add to RECOMMENDED_MODELS in server.py
• Write unit tests for parse_output()
• Optional: Write integration test with real CLI

HTTP Adapter Checklist

• Create adapters/your_adapter.py with build_request() and parse_response()
• Add to config.yaml with base_url, api_key, timeout
• Register in adapters/__init__.py (import + dict + export)
• Set environment variables for API keys
• Write unit tests with VCR cassettes
• Test with simple script before full deliberation

Common Commands

# Run unit tests for new adapter
pytest tests/unit/test_your_adapter.py -v

# Run with coverage
pytest tests/unit/test_your_adapter.py --cov=adapters.your_adapter

# Format and lint
black adapters/your_adapter.py && ruff check adapters/your_adapter.py

# Test integration (requires tool/service)
pytest tests/integration/ -v -m integration

# Record VCR cassette (first run, then commit)
pytest tests/unit/test_your_adapter.py -v

---

Additional Resources

• CLAUDE.md - Full project documentation with architecture details
• MCP Protocol - https://modelcontextprotocol.io/introduction
• Pydantic Docs - https://docs.pydantic.dev/latest/
• httpx Docs - https://www.python-httpx.org/
• VCR.py Docs - https://vcrpy.readthedocs.io/

---

This skill encodes the institutional knowledge for extending AI Counsel with new model integrations. Follow the patterns, write tests, and maintain backward compatibility.