feat(provider) add gemini provider

pyproject.toml

@@ -27,6 +27,7 @@ dependencies = [
     "aiofiles>=24.1.0",
     "paho-mqtt>=2.0.0",
     "matrix-nio>=0.24.0",
+    "google-genai>=1.67.0",
 ]
 
 [tool.setuptools.dynamic]

src/copaw/agents/model_factory.py

@@ -26,6 +26,13 @@
     AnthropicChatFormatter = None
     AnthropicChatModel = None
 
+try:
+    from agentscope.formatter import GeminiChatFormatter
+    from agentscope.model import GeminiChatModel
+except ImportError:  # pragma: no cover - compatibility fallback
+    GeminiChatFormatter = None
+    GeminiChatModel = None
+
 from .utils.tool_message_utils import _sanitize_tool_messages
 from ..providers import ProviderManager
 from ..providers.retry_chat_model import RetryChatModel
@@ -82,6 +89,8 @@ async def wrapper(
 }
 if AnthropicChatModel is not None and AnthropicChatFormatter is not None:
     _CHAT_MODEL_FORMATTER_MAP[AnthropicChatModel] = AnthropicChatFormatter
+if GeminiChatModel is not None and GeminiChatFormatter is not None:
+    _CHAT_MODEL_FORMATTER_MAP[GeminiChatModel] = GeminiChatFormatter
 
 
 def _get_formatter_for_chat_model(

src/copaw/app/routers/providers.py

@@ -14,7 +14,11 @@
 
 router = APIRouter(prefix="/models", tags=["models"])
 
-ChatModelName = Literal["OpenAIChatModel", "AnthropicChatModel"]
+ChatModelName = Literal[
+    "OpenAIChatModel",
+    "AnthropicChatModel",
+    "GeminiChatModel",
+]
 
 
 def get_provider_manager(request: Request) -> ProviderManager:

src/copaw/providers/gemini_provider.py

@@ -0,0 +1,130 @@
+# -*- coding: utf-8 -*-
+"""A Google Gemini provider implementation using AgentScope's native
+GeminiChatModel."""
+
+from __future__ import annotations
+
+from typing import Any, List
+
+from agentscope.model import ChatModelBase
+from google import genai
+from google.genai import errors as genai_errors
+from google.genai import types as genai_types
+
+from copaw.providers.provider import ModelInfo, Provider
+
+
+class GeminiProvider(Provider):
+    """Provider implementation for Google Gemini API."""
+
+    def _client(self, timeout: float = 5) -> Any:
+        return genai.Client(
+            api_key=self.api_key,
+            http_options=genai_types.HttpOptions(timeout=int(timeout * 1000)),
+        )
+
+    @staticmethod
+    def _normalize_models_payload(payload: Any) -> List[ModelInfo]:
+        models: List[ModelInfo] = []
+        for row in payload or []:
+            model_id = str(getattr(row, "name", "") or "").strip()
+
+            if not model_id:
+                continue
+
+            # Gemini API returns model names like "models/gemini-2.5-flash"
+            # Strip the "models/" prefix for cleaner IDs
+            if model_id.startswith("models/"):
+                model_id = model_id[len("models/") :]
+
+            display_name = str(
+                getattr(row, "display_name", "") or model_id,
+            ).strip()
+
+            if not display_name or display_name.startswith("models/"):
+                display_name = model_id
+
+            models.append(ModelInfo(id=model_id, name=display_name))
+
+        deduped: List[ModelInfo] = []
+        seen: set[str] = set()
+        for model in models:
+            if model.id in seen:
+                continue
+            seen.add(model.id)
+            deduped.append(model)
+        return deduped
+
+    async def check_connection(self, timeout: float = 5) -> tuple[bool, str]:
+        """Check if Google Gemini provider is reachable."""
+        try:
+            client = self._client(timeout=timeout)
+            # Use the async list models endpoint to verify connectivity
+            async for _ in await client.aio.models.list():
+                break
+            return True, ""
+        except genai_errors.APIError:
+            return (
+                False,
+                "Failed to connect to Google Gemini API. "
+                "Check your API key.",
+            )
+        except Exception:
+            return (
+                False,
+                "Unknown exception when connecting to Google Gemini API.",
+            )
+
+    async def fetch_models(self, timeout: float = 5) -> List[ModelInfo]:
+        """Fetch available models from Gemini API."""
+        try:
+            client = self._client(timeout=timeout)
+            payload = []
+            async for model in await client.aio.models.list():
+                payload.append(model)
+            models = self._normalize_models_payload(payload)
+            return models
+        except genai_errors.APIError:
+            return []
+        except Exception:
+            return []
+
+    async def check_model_connection(
+        self,
+        model_id: str,
+        timeout: float = 5,
+    ) -> tuple[bool, str]:
+        """Check if a specific Gemini model is reachable/usable."""
+        target = (model_id or "").strip()
+        if not target:
+            return False, "Empty model ID"
+
+        try:
+            client = self._client(timeout=timeout)
+            response = await client.aio.models.generate_content_stream(
+                model=target,
+                contents="ping",
+            )
+            async for _ in response:
+                break
+            return True, ""
+        except genai_errors.APIError:
+            return (
+                False,
+                f"Model '{model_id}' is not reachable or usable",
+            )
+        except Exception:
+            return (
+                False,
+                f"Unknown exception when connecting to model '{model_id}'",
+            )
+
+    def get_chat_model_instance(self, model_id: str) -> ChatModelBase:
+        from agentscope.model import GeminiChatModel
+
+        return GeminiChatModel(
+            model_name=model_id,
+            stream=True,
+            api_key=self.api_key,
+            generate_kwargs=self.generate_kwargs,
+        )

src/copaw/providers/provider_manager.py

@@ -21,6 +21,7 @@
 )
 from copaw.providers.openai_provider import OpenAIProvider
 from copaw.providers.anthropic_provider import AnthropicProvider
+from copaw.providers.gemini_provider import GeminiProvider
 from copaw.providers.ollama_provider import OllamaProvider
 from copaw.constant import SECRET_DIR
 from copaw.local_models import create_local_chat_model
@@ -92,6 +93,19 @@
 
 ANTHROPIC_MODELS: List[ModelInfo] = []
 
+GEMINI_MODELS: List[ModelInfo] = [
+    ModelInfo(id="gemini-3.1-pro-preview", name="Gemini 3.1 Pro Preview"),
+    ModelInfo(id="gemini-3-flash-preview", name="Gemini 3 Flash Preview"),
+    ModelInfo(
+        id="gemini-3.1-flash-lite-preview",
+        name="Gemini 3.1 Flash Lite Preview",
+    ),
+    ModelInfo(id="gemini-2.5-pro", name="Gemini 2.5 Pro"),
+    ModelInfo(id="gemini-2.5-flash", name="Gemini 2.5 Flash"),
+    ModelInfo(id="gemini-2.5-flash-lite", name="Gemini 2.5 Flash Lite"),
+    ModelInfo(id="gemini-2.0-flash", name="Gemini 2.0 Flash"),
+]
+
 PROVIDER_MODELSCOPE = OpenAIProvider(
     id="modelscope",
     name="ModelScope",
@@ -169,6 +183,17 @@
     freeze_url=True,
 )
 
+PROVIDER_GEMINI = GeminiProvider(
+    id="gemini",
+    name="Google Gemini",
+    base_url="https://generativelanguage.googleapis.com",
+    api_key_prefix="",
+    models=GEMINI_MODELS,
+    chat_model="GeminiChatModel",
+    freeze_url=True,
+    support_model_discovery=True,
+)
+
 PROVIDER_OLLAMA = OllamaProvider(
     id="ollama",
     name="Ollama",
@@ -244,6 +269,7 @@ def _init_builtins(self):
         self._add_builtin(PROVIDER_AZURE_OPENAI)
         self._add_builtin(PROVIDER_MINIMAX)
         self._add_builtin(PROVIDER_ANTHROPIC)
+        self._add_builtin(PROVIDER_GEMINI)
         self._add_builtin(PROVIDER_OLLAMA)
         self._add_builtin(PROVIDER_LMSTUDIO)
         self._add_builtin(PROVIDER_LLAMACPP)
@@ -454,6 +480,8 @@ def _provider_from_data(self, data: Dict) -> Provider:
 
         if provider_id == "anthropic" or chat_model == "AnthropicChatModel":
             return AnthropicProvider.model_validate(data)
+        if provider_id == "gemini" or chat_model == "GeminiChatModel":
+            return GeminiProvider.model_validate(data)
         if provider_id == "ollama":
             return OllamaProvider.model_validate(data)
         if data.get("is_local", False):