feat(claude_api): list models from Anthropic's live catalogue

ccproxy/plugins/claude_api/adapter.py

@@ -192,6 +192,10 @@ def _needs_anthropic_conversion(self, endpoint: str) -> bool:
         normalized = (endpoint or "").strip().lower()
         return normalized.startswith("/v1/chat/completions")
 
+    async def get_access_token(self) -> str:
+        """Return a usable OAuth access token for provider requests."""
+        return await self._resolve_access_token()
+
     async def _resolve_access_token(self) -> str:
         """Resolve a usable Claude API OAuth token from the token manager.
 

ccproxy/plugins/claude_api/config.py

@@ -41,6 +41,8 @@ class ClaudeAPISettings(ProviderConfig):
 
     # Feature flags
     include_sdk_content_as_xml: bool = False
+    # Query Anthropic for the current model list instead of using only static cards
+    live_models_enabled: bool = True
     support_openai_format: bool = True  # Support both Anthropic and OpenAI formats
 
     # System prompt injection mode

ccproxy/plugins/claude_api/live_models.py

@@ -0,0 +1,109 @@
+"""Live Anthropic model catalogue for the Claude API plugin."""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime
+from typing import Any
+
+import httpx
+
+from ccproxy.core.logging import get_plugin_logger
+
+
+logger = get_plugin_logger()
+
+MODELS_URL = "https://api.anthropic.com/v1/models"
+CACHE_TTL_SECONDS = 3600.0
+REQUEST_TIMEOUT_SECONDS = 15.0
+
+
+def _parse_created(created_at: Any) -> int:
+    """Convert an Anthropic RFC3339 timestamp into a Unix timestamp."""
+    if not isinstance(created_at, str) or not created_at:
+        return 0
+    try:
+        parsed = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
+    except ValueError:
+        return 0
+    return int(parsed.timestamp())
+
+
+class LiveModelCatalogue:
+    """Fetch Anthropic's current model list and cache it for a short while."""
+
+    def __init__(self, ttl_seconds: float = CACHE_TTL_SECONDS) -> None:
+        self._ttl = ttl_seconds
+        self._models: list[dict[str, Any]] = []
+        self._fetched_at = 0.0
+
+    def _is_fresh(self) -> bool:
+        return bool(self._models) and (time.monotonic() - self._fetched_at) < self._ttl
+
+    async def fetch(self, token: str) -> list[dict[str, Any]]:
+        """Return Anthropic's live models, keeping the last good list on failure."""
+        if self._is_fresh():
+            return self._models
+
+        headers = {
+            "authorization": f"Bearer {token}",
+            "anthropic-version": "2023-06-01",
+        }
+        try:
+            async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT_SECONDS) as client:
+                response = await client.get(MODELS_URL, headers=headers)
+                response.raise_for_status()
+                payload = response.json()
+        except Exception as exc:
+            logger.warning("live_model_catalogue_fetch_failed", error=str(exc))
+            return self._models
+
+        entries = payload.get("data") or []
+        models = [self._to_model(entry) for entry in entries if entry.get("id")]
+        if not models:
+            return self._models
+
+        self._models = models
+        self._fetched_at = time.monotonic()
+        return self._models
+
+    @staticmethod
+    def _to_model(entry: dict[str, Any]) -> dict[str, Any]:
+        """Convert an Anthropic model entry into an OpenAI-compatible model card."""
+        model: dict[str, Any] = {
+            "id": entry["id"],
+            "created": _parse_created(entry.get("created_at")),
+            "object": "model",
+            "owned_by": "anthropic",
+        }
+        for key in ("display_name", "line"):
+            value = entry.get(key)
+            if value:
+                model[key] = value
+        return model
+
+
+def merge_models(
+    live_models: list[dict[str, Any]], static_models: list[dict[str, Any]]
+) -> list[dict[str, Any]]:
+    """Merge live models with the static ones, live entries win."""
+    known = {model["id"] for model in live_models}
+    return live_models + [m for m in static_models if m.get("id") not in known]
+
+
+catalogue = LiveModelCatalogue()
+
+
+async def fetch_live_models(adapter: Any) -> list[dict[str, Any]]:
+    """Fetch the live catalogue with the adapter's OAuth token."""
+    getter = getattr(adapter, "get_access_token", None)
+    if getter is None:
+        return []
+
+    try:
+        token = await getter()
+    except Exception as exc:
+        logger.warning("live_model_token_unavailable", error=str(exc))
+        return []
+
+    return await catalogue.fetch(token)

ccproxy/plugins/claude_api/routes.py

@@ -24,6 +24,7 @@
 from ccproxy.streaming import DeferredStreaming
 
 from .config import ClaudeAPISettings
+from .live_models import fetch_live_models, merge_models
 
 
 if TYPE_CHECKING:
@@ -112,7 +113,15 @@ async def list_models(
     request: Request,
     auth: ConditionalAuthDep,
     config: ClaudeAPIConfigDep,
+    adapter: ClaudeAPIAdapterDep,
 ) -> dict[str, Any]:
-    """List available Claude models from configuration."""
-    models = [card.model_dump(mode="json") for card in config.models_endpoint]
-    return {"object": "list", "data": models}
+    """List available Claude models, merging Anthropic's live catalogue."""
+    static_models = [card.model_dump(mode="json") for card in config.models_endpoint]
+    if not config.live_models_enabled:
+        return {"object": "list", "data": static_models}
+
+    live_models = await fetch_live_models(adapter)
+    if not live_models:
+        return {"object": "list", "data": static_models}
+
+    return {"object": "list", "data": merge_models(live_models, static_models)}

tests/plugins/claude_api/unit/test_live_models.py

@@ -0,0 +1,108 @@
+"""Unit tests for the live Anthropic model catalogue."""
+
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from pytest_httpx import HTTPXMock
+
+from ccproxy.plugins.claude_api.live_models import (
+    LiveModelCatalogue,
+    fetch_live_models,
+    merge_models,
+)
+
+
+MODELS_URL = "https://api.anthropic.com/v1/models"
+
+
+class TestLiveModelCatalogue:
+    """Test the live model catalogue."""
+
+    async def test_fetch_converts_anthropic_entries(
+        self, httpx_mock: HTTPXMock
+    ) -> None:
+        """Anthropic entries become OpenAI-compatible model cards."""
+        httpx_mock.add_response(
+            url=MODELS_URL,
+            json={
+                "data": [
+                    {
+                        "id": "claude-opus-5-5",
+                        "created_at": "2026-05-01T12:00:00Z",
+                        "display_name": "Claude Opus 5.5",
+                        "line": "opus",
+                    }
+                ]
+            },
+        )
+
+        catalogue = LiveModelCatalogue()
+        models = await catalogue.fetch("test-token")
+
+        assert len(models) == 1
+        assert models[0]["id"] == "claude-opus-5-5"
+        assert models[0]["object"] == "model"
+        assert models[0]["owned_by"] == "anthropic"
+        assert models[0]["created"] == 1777636800
+        assert models[0]["display_name"] == "Claude Opus 5.5"
+        assert models[0]["line"] == "opus"
+
+        request = httpx_mock.get_requests()[0]
+        assert request.headers["authorization"] == "Bearer test-token"
+        assert request.headers["anthropic-version"] == "2023-06-01"
+
+    async def test_fetch_uses_cache_within_ttl(self, httpx_mock: HTTPXMock) -> None:
+        """A second fetch inside the TTL reuses the cached catalogue."""
+        httpx_mock.add_response(
+            url=MODELS_URL, json={"data": [{"id": "claude-opus-5-5"}]}
+        )
+
+        catalogue = LiveModelCatalogue()
+        await catalogue.fetch("test-token")
+        models = await catalogue.fetch("test-token")
+
+        assert [model["id"] for model in models] == ["claude-opus-5-5"]
+        assert len(httpx_mock.get_requests()) == 1
+
+    async def test_fetch_keeps_last_models_on_failure(
+        self, httpx_mock: HTTPXMock
+    ) -> None:
+        """An upstream failure keeps the last successful catalogue."""
+        httpx_mock.add_response(
+            url=MODELS_URL, json={"data": [{"id": "claude-opus-5-5"}]}
+        )
+        catalogue = LiveModelCatalogue(ttl_seconds=0.0)
+        await catalogue.fetch("test-token")
+
+        httpx_mock.reset()
+        httpx_mock.add_response(url=MODELS_URL, status_code=500)
+        models = await catalogue.fetch("test-token")
+
+        assert [model["id"] for model in models] == ["claude-opus-5-5"]
+
+    async def test_fetch_without_token_returns_nothing(self) -> None:
+        """Without a usable token the live catalogue stays empty."""
+        adapter = Mock()
+        adapter.get_access_token = AsyncMock(side_effect=RuntimeError("no credentials"))
+
+        assert await fetch_live_models(adapter) == []
+
+
+class TestMergeModels:
+    """Test merging live and static model cards."""
+
+    def test_live_models_win(self) -> None:
+        """Live entries replace static ones with the same id."""
+        live = [{"id": "claude-sonnet-4-6", "created": 1}]
+        static = [
+            {"id": "claude-sonnet-4-6", "created": 0},
+            {"id": "claude-3-5-sonnet-20241022", "created": 0},
+        ]
+
+        merged = merge_models(live, static)
+
+        assert [model["id"] for model in merged] == [
+            "claude-sonnet-4-6",
+            "claude-3-5-sonnet-20241022",
+        ]
+        assert merged[0]["created"] == 1