The Adapter Pattern — Wrapping Legacy CLI for AI Agent Consumption

English title: API-First, CLI-Second: The Adapter Pattern

Mở đầu

Khi GitHub ra mắt gh — CLI chính thức của họ — vào năm 2020, codebase có một điều đặc biệt: tất cả business logic nằm trong một tầng riêng biệt. CLI commands chỉ làm một việc: parse argument, gọi adapter, format output. API calls, auth, pagination, error handling — tất cả nằm trong adapter.

Đó không phải ngẫu nhiên. GitHub CLI team đến từ oclif ecosystem, nơi pattern này đã được dùng từ Heroku CLI. Stripe CLI dùng pattern tương tự. kubectl, Vercel CLI, AWS CLI v2 — tất cả đều có một tầng trung gian giữa raw API và CLI interface.

Pattern này có tên: Adapter Pattern (trong ngữ cảnh CLI).

Bài này đi vào chi tiết tại sao pattern này tồn tại, nó giải quyết vấn đề gì, và cách implement từ đầu bằng Python.

1. Vấn đề mà adapter giải quyết

Hãy tưởng tượng bạn viết CLI trực tiếp gọi REST API:

# ❌ Anti-pattern: CLI command gọi thẳng API
@app.command()
def list_users(token: str = typer.Option(..., envvar="API_TOKEN")):
    response = httpx.get(
        "https://api.example.com/v1/users",
        headers={"Authorization": f"Bearer {token}"},
        params={"page": 1, "per_page": 100}
    )
    if response.status_code == 401:
        print("Unauthorized")
        raise SystemExit(1)
    if response.status_code == 429:
        print("Rate limited")
        raise SystemExit(1)
 
    data = response.json()
    # Handle pagination manually in every command...
    while data.get("has_next_page"):
        next_response = httpx.get(
            "https://api.example.com/v1/users",
            headers={"Authorization": f"Bearer {token}"},
            params={"page": data["page"] + 1, "per_page": 100}
        )
        data["items"].extend(next_response.json()["items"])
 
    for user in data["items"]:
        print(f"{user['name']:10s} {user['email']}")

Khi bạn có 50 commands, đoạn auth + pagination + error handling này sẽ xuất hiện 50 lần. Khi API thay đổi base URL, thêm rate limiting header, hoặc đổi format pagination — bạn sửa 50 chỗ.

Adapter pattern tách bạch hai mối lo khác nhau:

CLI layer: Parse arguments, format output, exit codes
Adapter layer: Auth, HTTP, pagination, retry, error mapping

2. Anatomy của một adapter

Adapter là class Python thuần — không phụ thuộc vào CLI framework. Nó có thể được test độc lập, và sau này wrap thành MCP server mà không cần đụng vào CLI code.

# adapters/user_adapter.py
import httpx
from typing import Iterator
from dataclasses import dataclass
 
 
@dataclass
class User:
    id: int
    name: str
    email: str
    role: str
 
 
class UserAdapter:
    """Adapter giữa GitHub-style User API và CLI/agent consumers."""
 
    BASE_URL = "https://api.example.com/v1"
    PAGE_SIZE = 100
 
    def __init__(self, token: str, timeout: int = 30):
        self._client = httpx.Client(
            base_url=self.BASE_URL,
            headers={
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
            },
            timeout=timeout,
        )
 
    def _request(self, method: str, path: str, **kwargs) -> dict:
        """Central HTTP method với error mapping."""
        try:
            resp = self._client.request(method, path, **kwargs)
        except httpx.TimeoutException:
            raise AdapterError("timeout", "Request timed out", exit_code=1)
        except httpx.ConnectError:
            raise AdapterError("network", "Cannot connect to API", exit_code=1)
 
        # Map HTTP status → domain errors
        if resp.status_code == 401:
            raise AdapterError("auth", "Invalid or expired token", exit_code=4)
        if resp.status_code == 403:
            raise AdapterError("permission", "Insufficient permissions", exit_code=4)
        if resp.status_code == 404:
            raise AdapterError("not_found", resp.json().get("message", "Not found"), exit_code=3)
        if resp.status_code == 409:
            raise AdapterError("conflict", "Resource already exists", exit_code=5)
        if resp.status_code == 429:
            retry_after = resp.headers.get("Retry-After", "60")
            raise AdapterError("rate_limited", f"Rate limited. Retry after {retry_after}s", exit_code=1)
        if not resp.is_success:
            raise AdapterError("api_error", f"API error {resp.status_code}", exit_code=1)
 
        return resp.json()
 
    def list_users(self, role: str | None = None) -> Iterator[User]:
        """List users với auto-pagination."""
        page = 1
        while True:
            params = {"page": page, "per_page": self.PAGE_SIZE}
            if role:
                params["role"] = role
 
            data = self._request("GET", "/users", params=params)
            items = data.get("items", [])
 
            for item in items:
                yield User(
                    id=item["id"],
                    name=item["name"],
                    email=item["email"],
                    role=item.get("role", "member"),
                )
 
            if not data.get("has_next_page") or len(items) < self.PAGE_SIZE:
                break
            page += 1
 
    def get_user(self, user_id: int) -> User:
        """Get single user by ID."""
        data = self._request("GET", f"/users/{user_id}")
        return User(
            id=data["id"],
            name=data["name"],
            email=data["email"],
            role=data.get("role", "member"),
        )
 
    def ensure_user(self, name: str, email: str, role: str = "member") -> tuple[User, str]:
        """Create user nếu chưa tồn tại. Idempotent. Returns (user, action)."""
        try:
            # Try to find existing
            data = self._request("GET", "/users", params={"email": email})
            if data.get("items"):
                user_data = data["items"][0]
                return User(**user_data), "already_exists"
        except AdapterError as e:
            if e.code != "not_found":
                raise
 
        # Create new
        data = self._request("POST", "/users", json={"name": name, "email": email, "role": role})
        return User(id=data["id"], name=name, email=email, role=role), "created"
 
    def close(self):
        self._client.close()
 
    def __enter__(self):
        return self
 
    def __exit__(self, *args):
        self.close()
 
 
class AdapterError(Exception):
    """Lỗi từ adapter — CLI sẽ catch và map sang exit code."""
    def __init__(self, code: str, message: str, exit_code: int = 1):
        super().__init__(message)
        self.code = code
        self.exit_code = exit_code

Adapter xử lý 4 thứ:

Auth: Token inject một lần trong constructor, không lặp trong mỗi method
Pagination: Iterator pattern — caller không biết và không cần biết có nhiều trang
Error mapping: HTTP status codes → AdapterError với semantic code và exit code
Retry/timeout: Tập trung tại _request(), không scatter khắp nơi

3. CLI layer — mỏng và sạch

Khi đã có adapter, CLI commands trở nên rất mỏng:

# cli/users.py
import json
import sys
import os
import typer
from adapters.user_adapter import UserAdapter, AdapterError
from dataclasses import asdict
 
app = typer.Typer(help="User management commands")
 
 
def get_adapter() -> UserAdapter:
    """Factory: load token từ env, tạo adapter."""
    token = os.environ.get("API_TOKEN")
    if not token:
        print(json.dumps({"error": "missing_token", "message": "Set API_TOKEN env var"}), file=sys.stderr)
        raise SystemExit(2)
    return UserAdapter(token=token)
 
 
def handle_adapter_error(e: AdapterError) -> None:
    """Convert AdapterError → stderr JSON + exit code."""
    print(json.dumps({"error": e.code, "message": str(e)}), file=sys.stderr)
    raise SystemExit(e.exit_code)
 
 
@app.command()
def list_users(
    role: str | None = typer.Option(None, help="Filter by role: admin, member, viewer"),
    json_output: bool = typer.Option(False, "--json"),
):
    """List users. Supports pagination automatically."""
    try:
        with get_adapter() as adapter:
            users = list(adapter.list_users(role=role))
    except AdapterError as e:
        handle_adapter_error(e)
 
    if json_output:
        print(json.dumps([asdict(u) for u in users], indent=2))
    else:
        for u in users:
            print(f"  {u.name:15s} {u.email:25s} {u.role}")
 
 
@app.command()
def get_user(
    user_id: int = typer.Argument(..., help="User ID"),
    json_output: bool = typer.Option(False, "--json"),
):
    """Get user by ID."""
    try:
        with get_adapter() as adapter:
            user = adapter.get_user(user_id)
    except AdapterError as e:
        handle_adapter_error(e)
 
    if json_output:
        print(json.dumps(asdict(user), indent=2))
    else:
        print(f"  id:    {user.id}")
        print(f"  name:  {user.name}")
        print(f"  email: {user.email}")
        print(f"  role:  {user.role}")
 
 
@app.command()
def ensure_user(
    name: str = typer.Option(..., help="User full name"),
    email: str = typer.Option(..., help="User email"),
    role: str = typer.Option("member", help="Role: admin, member, viewer"),
    json_output: bool = typer.Option(False, "--json"),
):
    """Create user if not exists. Idempotent — safe to run multiple times."""
    try:
        with get_adapter() as adapter:
            user, action = adapter.ensure_user(name=name, email=email, role=role)
    except AdapterError as e:
        handle_adapter_error(e)
 
    result = {"action": action, "user": asdict(user)}
    if json_output:
        print(json.dumps(result, indent=2))
    else:
        icon = "✓" if action == "already_exists" else "+"
        print(f"  {icon} [{action}] {user.name} <{user.email}>")

So sánh với anti-pattern ở đầu bài: CLI commands bây giờ không biết gì về HTTP, không biết gì về pagination, không biết gì về auth. Chúng chỉ:

Gọi get_adapter() để lấy adapter
Gọi method trên adapter
Catch AdapterError và map sang exit code
Format output

4. Tại sao pattern này quan trọng cho agent

Khi agent gọi CLI, nó không quan tâm đến HTTP. Nó quan tâm đến:

Structured output: Adapter đảm bảo data shape nhất quán
Exit codes: AdapterError.exit_code → CLI exit code, không cần agent đọc error message
Idempotency: ensure_user safe to retry — agent có thể retry khi timeout mà không tạo duplicate

Nhưng quan trọng hơn: adapter có thể reuse cho nhiều consumers.

REST API
    │
    ▼
UserAdapter
    ├── CLI (subprocess.run)          ← agent gọi qua subprocess
    ├── MCP Server                    ← agent gọi qua tool_call
    └── Python SDK                    ← human developer import trực tiếp

Khi bạn muốn expose tool qua MCP (bài 7), bạn không rewrite business logic — bạn wrap adapter:

# mcp_server.py — bài 7 sẽ đi sâu vào phần này
from mcp import Server
from adapters.user_adapter import UserAdapter
 
server = Server("user-management")
 
@server.tool("list_users")
async def mcp_list_users(role: str | None = None) -> list[dict]:
    with UserAdapter(token=os.environ["API_TOKEN"]) as adapter:
        return [asdict(u) for u in adapter.list_users(role=role)]

Adapter = business logic. CLI, MCP, SDK = delivery mechanism. Separation of concerns.

5. Workflow: Từ API spec đến adapter

Quy trình thực tế khi xây CLI từ REST API:

Bước 1: Map API endpoints → adapter methods

GET  /users         → list_users(role?) → Iterator[User]
GET  /users/{id}    → get_user(id) → User
POST /users         → create_user(name, email, role) → User
PUT  /users/{id}    → update_user(id, **kwargs) → User
DELETE /users/{id}  → delete_user(id) → None

Bước 2: Define domain errors

ERROR_MAP = {
    401: AdapterError("auth", exit_code=4),
    403: AdapterError("permission", exit_code=4),
    404: AdapterError("not_found", exit_code=3),
    409: AdapterError("conflict", exit_code=5),
    429: AdapterError("rate_limited", exit_code=1),
}

Bước 3: Define output dataclasses

Output types cố định → CLI format nhất quán → agent parse nhất quán.

Bước 4: Implement pagination strategy

API khác nhau dùng pagination khác nhau:

Offset-based: ?page=1&per_page=100 (GitHub-style)
Cursor-based: ?after=cursor_token (Stripe-style)
Link header: Link: <next_url>; rel="next" (GitHub API v3)

Adapter handle tất cả — CLI chỉ gọi iterator, không biết pagination strategy là gì.

Bước 5: Test adapter độc lập

# tests/unit/test_user_adapter.py
import pytest
import respx
import httpx
from adapters.user_adapter import UserAdapter, AdapterError
 
@respx.mock
def test_list_users_pagination():
    # Mock page 1
    respx.get("https://api.example.com/v1/users").mock(
        side_effect=[
            httpx.Response(200, json={"items": [{"id": 1, "name": "Alice", "email": "a@x.com", "role": "admin"}], "has_next_page": True}),
            httpx.Response(200, json={"items": [{"id": 2, "name": "Bob", "email": "b@x.com", "role": "member"}], "has_next_page": False}),
        ]
    )
 
    adapter = UserAdapter(token="test-token")
    users = list(adapter.list_users())
    assert len(users) == 2
    assert users[0].name == "Alice"
    assert users[1].name == "Bob"
 
@respx.mock
def test_auth_error_maps_to_exit_code_4():
    respx.get("https://api.example.com/v1/users/999").mock(
        return_value=httpx.Response(401, json={"message": "Unauthorized"})
    )
 
    adapter = UserAdapter(token="bad-token")
    with pytest.raises(AdapterError) as exc_info:
        adapter.get_user(999)
 
    assert exc_info.value.exit_code == 4
    assert exc_info.value.code == "auth"

Test adapter — không test CLI. Điều này cho phép bạn swap CLI framework (Typer → Click → Cobra) mà không rewrite tests.

6. Ứng dụng trong AI-centric engineering

GitHub CLI (gh) có hơn 200 commands nhưng chỉ một handful of adapters: api.go, github.go, browser.go. Mỗi adapter xử lý một domain. CLI commands là thin wrappers.

Stripe CLI dùng pattern tương tự: stripe.Client là adapter, commands chỉ gọi methods trên client.

Kết quả: khi Stripe thêm idempotency keys vào API, họ sửa một chỗ trong adapter — không phải 200 chỗ trong 200 commands.

Với AI agent, pattern này quan trọng hơn vì agent retry nhiều hơn người. Nếu create_user không idempotent, agent timeout lần đầu và retry sẽ tạo duplicate. ensure_user trong adapter xử lý điều này — agent gọi bao nhiêu lần cũng được.

Tóm tắt pattern:

API Spec
  → Adapter (auth, pagination, error mapping, idempotency)
    → CLI (thin: parse args, call adapter, format output)
      → Agent (subprocess.run, parse JSON, check exit code)

Bài tiếp, chúng ta sẽ đi vào bước trước adapter: extract API spec thành structured metadata — JSON document mô tả 439 endpoints, là input cho code generation pipeline tạo ra adapter và commands tự động.

Bài tiếp

Bài 4: From API Spec to Structured Metadata — Khi bạn có 439 API endpoints, viết tay adapter và commands mất tuần. Extract metadata từ OpenAPI spec (hoặc HTML docs) → JSON "nguồn sự thật" → làm input cho codegen pipeline. Code ví dụ: scraping, parsing, normalization.