feat: fetch provider models from the UI and one-click add to the composer picker

coworker/providers/__init__.py

@@ -17,6 +17,7 @@
     build_provider_client,
     descriptor_configured,
     detect_provider,
+    fetch_provider_models,
     get_descriptor,
     provider_descriptors,
     provider_names,

coworker/providers/_models_candidates.py

@@ -0,0 +1,121 @@
+"""Model-fetch candidate URL generation — ported from cc-switch's `model_fetch.rs`.
+
+`build_models_candidates` generates the list of URLs to try when fetching a provider's
+model list, handling the same edge cases cc-switch does: known compat suffixes, version
+segment endings, and full-URL parsing.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Optional
+
+# Suffixes that indicate a _compatibility_ sub-path — the real `/v1/models` endpoint
+# lives at the root, not under this path. Ported from cc-switch's `KNOWN_COMPAT_SUFFIXES`.
+KNOWN_COMPAT_SUFFIXES: list[str] = [
+    "/api/claudecode",
+    "/api/anthropic",
+    "/apps/anthropic",
+    "/api/coding",
+    "/claudecode",
+    "/anthropic",
+    "/step_plan",
+    "/coding",
+    "/claude",
+]
+
+# Version-segment patterns: if the URL ends with one of these, the `/models` endpoint
+# lives at the same level (i.e. strip the version segment, not append).
+_VERSION_SEGMENTS = {"/v1", "/v4", "/v2", "/v1beta", "/v1beta1", "/v1beta2"}
+
+
+def _strip_compat_suffix(url: str) -> Optional[str]:
+    """If `url` ends with a known compat suffix, return the URL without it."""
+    for suffix in KNOWN_COMPAT_SUFFIXES:
+        if url.endswith(suffix):
+            return url[: -len(suffix)]
+    return None
+
+
+def build_models_candidates(
+    base_url: str,
+    *,
+    is_full_url: bool = False,
+    models_url_override: Optional[str] = None,
+) -> list[str]:
+    """Build the list of candidate URLs to try when fetching models.
+
+    Mirrors cc-switch's `build_models_url_candidates` (model_fetch.rs:139-200).
+
+    Args:
+        base_url: The provider's base URL (e.g. ``https://api.deepseek.com``).
+        is_full_url: If True, treat ``base_url`` as a full request URL; extract the path
+            prefix and append ``/v1/models``.
+        models_url_override: If set, use ONLY this URL (e.g. DeepSeek's ``/models``
+            lives at the root path).
+
+    Returns:
+        A list of candidate URLs, in order of preference.
+    """
+    candidates: list[str] = []
+
+    # 1. Explicit override — use as-is, no further guessing.
+    if models_url_override:
+        return [models_url_override]
+
+    url = base_url.strip().rstrip("/")
+
+    # 2. Full-URL mode: extract the path prefix (e.g. ``/api/paas/v4`` from
+    #    ``https://api.z.ai/api/paas/v4``) and append ``/v1/models``.
+    if is_full_url:
+        # Extract the path from the URL (everything after the host)
+        match = re.match(r"https?://[^/]+(/.*)", url)
+        if match:
+            path_prefix = match.group(1).rstrip("/")
+            candidates.append(url + "/v1/models")
+            # Also try: if the path prefix looks like a version segment, try
+            # replacing it with ``/v1/models``.
+            # cc-switch does: `format!("{path_prefix}/v1/models")`
+            # Actually cc-switch does: `format!("{url}/v1/models")` for is_full_url
+            # Let me re-read the Rust code...
+            # The Rust code:
+            #   if is_full_url {
+            #       candidates.push(format!("{url}/v1/models"));
+            #       // also try stripping the last path segment
+            #       if let Some(parent) = Path::new(path_prefix).parent() {
+            #           candidates.push(format!("{}{}/v1/models", origin, parent.display()));
+            #       }
+            #   }
+            # So it first tries {url}/v1/models, then {origin}{parent}/v1/models
+            candidates.append(url + "/v1/models")
+            # Also try parent path
+            parts = path_prefix.rsplit("/", 1)
+            if len(parts) > 1:
+                parent = parts[0]
+                candidates.append(url[: len(url) - len(path_prefix)] + parent + "/v1/models")
+        else:
+            candidates.append(url + "/v1/models")
+        # Also try the root path
+        candidates.append(url + "/models")
+        return candidates
+
+    # 3. Normal mode: start with ``{base}/v1/models``
+    candidates.append(url + "/v1/models")
+
+    # 4. If the URL ends with a version segment (``/v1``, ``/v4``, etc.), also try
+    #    ``{base}/models`` (i.e. strip the version segment rather than appending).
+    for vs in _VERSION_SEGMENTS:
+        if url.endswith(vs):
+            candidates.append(url + "/models")
+            break
+
+    # 5. Check for known compat suffixes — strip them and retry with the root.
+    stripped = _strip_compat_suffix(url)
+    if stripped:
+        candidates.append(stripped + "/v1/models")
+        candidates.append(stripped + "/models")
+
+    # 6. Last resort: try the bare root path
+    candidates.append(url + "/models")
+
+    return candidates

coworker/providers/registry.py

@@ -864,3 +864,210 @@ def verify_provider_key(
             "error": "Reached the server, but no OpenAI-compatible /v1 API there.",
         }
     return {"ok": False, "error": f"{d.title} returned HTTP {resp.status_code}."}
+
+
+def fetch_provider_models(
+    name: str,
+    *,
+    api_key: Optional[str] = None,
+    base_url: Optional[str] = None,
+    fields: Optional[dict[str, Any]] = None,
+    timeout: float = 10.0,
+) -> dict[str, Any]:
+    """Fetch the list of models available from a provider.
+
+    Like ``verify_provider_key`` but returns the parsed model list on success. Uses
+    candidate URL generation (ported from cc-switch) to handle edge cases: compat
+    suffixes, version segments, and custom endpoints.
+
+    Returns:
+        ``{"ok": True, "models": [{"id": str, "owned_by": str | None}, ...]}`` on success,
+        or ``{"ok": False, "error": str}`` on failure.
+    """
+    from ._models_candidates import build_models_candidates
+
+    import httpx
+
+    d = _BY_NAME.get(name) or _BY_NAME["openai"]
+    key = (api_key or "").strip()
+
+    # Cloud providers (Bedrock, Vertex) — use their existing verify logic
+    if name == "bedrock":
+        result = _verify_bedrock(fields or {}, timeout)
+        if not result.get("ok"):
+            return result
+        # Bedrock: list models via boto3
+        from .bedrock_provider import _session_kwargs
+
+        def get(key: str) -> Optional[str]:
+            return ((fields or {}).get(key) or "").strip() or None
+
+        try:
+            import boto3
+            from botocore.config import Config
+
+            method = get("auth_method") or "api_key"
+            session_kwargs: dict[str, Any] = {}
+            if method == "api_key":
+                if get("bedrock_api_key"):
+                    os.environ["AWS_BEARER_TOKEN_BEDROCK"] = get("bedrock_api_key")
+            elif method == "profile":
+                session_kwargs = _session_kwargs(get("aws_profile"), None, None, None)
+            else:  # iam
+                session_kwargs = _session_kwargs(
+                    None, get("aws_access_key_id"), get("aws_secret_access_key"), get("aws_session_token")
+                )
+            session = boto3.session.Session(**session_kwargs)
+            client = session.client(
+                "bedrock",
+                region_name=get("region"),
+                config=Config(connect_timeout=timeout, read_timeout=timeout),
+            )
+            models = client.list_foundation_models()
+            data = [
+                {"id": m["modelId"], "owned_by": m.get("providerName")}
+                for m in models.get("modelSummaries", [])
+            ]
+            return {"ok": True, "models": data}
+        except Exception as exc:
+            return {"ok": False, "error": f"Couldn't list Bedrock models ({exc.__class__.__name__})."}
+
+    if name == "vertex":
+        result = _verify_vertex(fields or {}, timeout)
+        if not result.get("ok"):
+            return result
+        # Vertex: list models via the Vertex AI API
+        project = ((fields or {}).get("project") or "").strip()
+        location = ((fields or {}).get("location") or "").strip()
+        method = ((fields or {}).get("auth_method") or "").strip() or "adc"
+        try:
+            if method == "api_key":
+                vkey = ((fields or {}).get("vertex_api_key") or "").strip()
+                resp = httpx.get(
+                    f"https://aiplatform.googleapis.com/v1/projects/{project}/locations/{location}/models",
+                    headers={"x-goog-api-key": vkey},
+                    timeout=timeout,
+                )
+            else:
+                from .vertex_provider import _regional_host, load_credentials
+
+                creds = None
+                if method == "service_account":
+                    creds = load_credentials((fields or {}).get("service_account_json"))
+                if creds is None:
+                    import google.auth
+                    creds, _ = google.auth.default(
+                        scopes=["https://www.googleapis.com/auth/cloud-platform"]
+                    )
+                from google.auth.transport.requests import Request
+                creds.refresh(Request())
+                resp = httpx.get(
+                    f"https://{_regional_host(location)}/v1/projects/{project}/locations/{location}/models",
+                    headers={"Authorization": f"Bearer {creds.token}"},
+                    timeout=timeout,
+                )
+            if resp.status_code < 300:
+                data = resp.json().get("models", [])
+                models = [{"id": m.get("name", ""), "owned_by": None} for m in data]
+                return {"ok": True, "models": models}
+            return {"ok": False, "error": f"Vertex AI returned HTTP {resp.status_code}."}
+        except Exception as exc:
+            return {"ok": False, "error": f"Couldn't list Vertex models ({exc.__class__.__name__})."}
+
+    # Native API providers + generic OpenAI-compatible
+    try:
+        if name == "anthropic":
+            candidates = build_models_candidates(
+                "https://api.anthropic.com",
+                is_full_url=False,
+            )
+            headers = {"x-api-key": key, "anthropic-version": "2023-06-01"}
+            resp = None
+            for url in candidates:
+                try:
+                    resp = httpx.get(url, headers=headers, timeout=timeout)
+                    if resp.status_code < 300:
+                        break
+                except Exception:
+                    continue
+            if resp is None or resp.status_code >= 300:
+                return {"ok": False, "error": "Couldn't reach Anthropic API."}
+            data = resp.json().get("data", [])
+            models = [{"id": m["id"], "owned_by": m.get("owned_by")} for m in data]
+            return {"ok": True, "models": models}
+
+        elif name == "gemini":
+            candidates = build_models_candidates(
+                "https://generativelanguage.googleapis.com/v1beta",
+                is_full_url=False,
+            )
+            resp = None
+            for url in candidates:
+                try:
+                    resp = httpx.get(url, params={"key": key}, timeout=timeout)
+                    if resp.status_code < 300:
+                        break
+                except Exception:
+                    continue
+            if resp is None or resp.status_code >= 300:
+                return {"ok": False, "error": "Couldn't reach Gemini API."}
+            data = resp.json().get("models", [])
+            models = [{"id": m.get("name", "").split("/")[-1], "owned_by": None} for m in data]
+            return {"ok": True, "models": models}
+
+        elif name == "ollama":
+            base = _normalize_ollama_url(base_url)
+            # Ollama's /v1/models returns OpenAI-compatible model list
+            candidates = build_models_candidates(base.rstrip("/v1"), is_full_url=False)
+            resp = None
+            for url in candidates:
+                try:
+                    resp = httpx.get(url, timeout=timeout)
+                    if resp.status_code < 300:
+                        break
+                except Exception:
+                    continue
+            if resp is None or resp.status_code >= 300:
+                return {"ok": False, "error": "Couldn't reach Ollama."}
+            data = resp.json().get("data", [])
+            models = [{"id": m["id"], "owned_by": m.get("owned_by")} for m in data]
+            return {"ok": True, "models": models}
+
+        else:  # openai + any OpenAI-compatible endpoint
+            default_base = next(
+                (f.default for f in d.fields if f.key == "base_url" and f.default), ""
+            )
+            base = (
+                (base_url or "").strip().rstrip("/")
+                or default_base.rstrip("/")
+                or "https://api.openai.com/v1"
+            )
+            # Determine if this is a full URL (has a non-trivial path like /api/paas/v4)
+            from urllib.parse import urlparse
+
+            parsed = urlparse(base)
+            is_full = bool(parsed.path) and parsed.path not in ("", "/", "/v1")
+            candidates = build_models_candidates(base, is_full_url=is_full)
+            resp = None
+            for url in candidates:
+                try:
+                    resp = httpx.get(
+                        url,
+                        headers={"Authorization": f"Bearer {key}"},
+                        timeout=timeout,
+                    )
+                    if resp.status_code < 300:
+                        break
+                except Exception:
+                    continue
+            if resp is None or resp.status_code >= 300:
+                return {"ok": False, "error": f"Couldn't reach {d.title}."}
+            data = resp.json().get("data", [])
+            models = [{"id": m["id"], "owned_by": m.get("owned_by")} for m in data]
+            return {"ok": True, "models": models}
+
+    except Exception as exc:
+        return {
+            "ok": False,
+            "error": f"Couldn't reach {d.title} ({exc.__class__.__name__}).",
+        }

coworker/server/app.py

@@ -1382,6 +1382,13 @@ async def providers_verify(body: dict) -> dict[str, Any]:
             manager.verify_provider, name, (body or {}).get("fields")
         )
 
+    @app.post("/v1/providers/{name}/models")
+    async def providers_fetch_models(name: str, body: dict) -> dict[str, Any]:
+        """Fetch the model list from a provider (live read-only call, does NOT persist)."""
+        return await asyncio.to_thread(
+            manager.fetch_models, name, (body or {}).get("fields")
+        )
+
     # -- settings (model API key) -----------------------------------------------
     @app.get("/v1/settings")
     def settings_get() -> dict[str, Any]: