feat(ai): integrate Gemma 4 with Native Function Calling and multi-backend support

- Add GemmaModel enum (2B/4B/26B/31B) and AIBackend enum (Google AI Studio/LM Studio/Ollama)

- Implement GEMMA_TOOLS for Native Function Calling (analyze_schema, generate_column_values)

- Add _try_tool_calling() with graceful fallback to JSON mode then plain text

- Add LM Studio backend support for local Edge deployment

- Add 3 MCP tools: sqlseed_gemma4_analyze, sqlseed_gemma4_agent_fill, sqlseed_list_gemma_models

- Add google-generativeai dependency

Author: sunbos

plugins/mcp-server-sqlseed/src/mcp_server_sqlseed/server.py

@@ -8,12 +8,12 @@
 import yaml
 from mcp.server.fastmcp import FastMCP
 
-from sqlseed.config.models import GeneratorConfig
+from sqlseed.config.models import ColumnConfig, GeneratorConfig
 from sqlseed.core.orchestrator import DataOrchestrator
 
 try:
     from sqlseed_ai.analyzer import SchemaAnalyzer
-    from sqlseed_ai.config import AIConfig
+    from sqlseed_ai.config import AIBackend, AIConfig, GemmaModel
     from sqlseed_ai.refiner import AiConfigRefiner, AISuggestionFailedError
 
     _AI_AVAILABLE = True
@@ -188,3 +188,155 @@ def sqlseed_execute_fill(
             "elapsed": result.elapsed,
             "errors": result.errors,
         }
+
+
+@mcp.tool()
+def sqlseed_gemma4_analyze(
+    db_path: str,
+    table_name: str,
+    model: str | None = None,
+    backend: str | None = None,
+) -> dict[str, Any]:
+    """Analyze a database table schema using Gemma 4 with native function calling.
+
+    This tool leverages Gemma 4's built-in tool use capability to analyze
+    table structure and recommend data generation configurations. It demonstrates
+    Gemma 4's Native Function Calling feature for the AI Agent track.
+
+    Supported backends: google_ai_studio (default), ollama, openai_compat.
+    Supported models: gemma-4-26b-it (default), gemma-4-31b-it, gemma-4-4b-it, gemma-4-2b-it.
+    """
+    if not _AI_AVAILABLE:
+        return {"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai"}
+
+    db_path = _validate_db_path(db_path)
+    with DataOrchestrator(db_path) as orch:
+        _validate_table_name(table_name, orch.get_table_names())
+        schema_ctx = orch.get_schema_context(table_name)
+
+    # Build AIConfig with Gemma 4 defaults
+    ai_config = AIConfig.from_env()
+    if model:
+        ai_config.model = model
+    if backend:
+        try:
+            ai_config.backend = AIBackend(backend)
+        except ValueError:
+            return {"error": f"Invalid backend: {backend}. Use: google_ai_studio, ollama, openai_compat"}
+    ai_config.resolve_model()
+
+    analyzer = SchemaAnalyzer(config=ai_config)
+    result = analyzer.analyze_table_from_ctx(**_serialize_schema_context(schema_ctx))
+
+    if result is None:
+        return {"error": "Gemma 4 analysis returned no result. Check API key and model availability."}
+
+    return {
+        "model": ai_config.model,
+        "backend": ai_config.backend.value,
+        "table_name": table_name,
+        "config": result,
+    }
+
+
+@mcp.tool()
+def sqlseed_gemma4_agent_fill(
+    db_path: str,
+    table_name: str,
+    count: int = 1000,
+    model: str | None = None,
+    backend: str | None = None,
+    max_retries: int = 3,
+) -> dict[str, Any]:
+    """End-to-end AI Agent: Gemma 4 analyzes schema → generates config → fills data.
+
+    This is a complete Agent workflow that demonstrates Gemma 4's Native Function
+    Calling capability for the AI Agent track:
+    1. Inspect schema (Tool Calling: analyze_schema)
+    2. Generate data configuration (self-correction loop)
+    3. Execute data fill
+
+    The agent uses Gemma 4's tool use to understand schema semantics and
+    produce appropriate data generation rules automatically.
+    """
+    if not _AI_AVAILABLE:
+        return {"error": "sqlseed-ai plugin not installed. Install with: pip install sqlseed-ai"}
+
+    db_path = _validate_db_path(db_path)
+
+    # Step 1: AI analysis with self-correction
+    ai_config = AIConfig.from_env()
+    if model:
+        ai_config.model = model
+    if backend:
+        try:
+            ai_config.backend = AIBackend(backend)
+        except ValueError:
+            return {"error": f"Invalid backend: {backend}. Use: google_ai_studio, ollama, openai_compat"}
+    ai_config.resolve_model()
+
+    analyzer = SchemaAnalyzer(config=ai_config)
+    refiner = AiConfigRefiner(analyzer, db_path)
+
+    try:
+        ai_result = refiner.generate_and_refine(
+            table_name=table_name,
+            max_retries=max_retries,
+        )
+    except AISuggestionFailedError as e:
+        return {"error": f"AI suggestion failed: {e}", "model": ai_config.model}
+    except (ValueError, RuntimeError, OSError) as e:
+        return {"error": f"Error: {e}", "model": ai_config.model}
+
+    if not ai_result:
+        return {"error": "No AI suggestions available", "model": ai_config.model}
+
+    # Step 2: Execute fill with AI-generated config
+    with DataOrchestrator(db_path) as orch:
+        _validate_table_name(table_name, orch.get_table_names())
+
+        column_configs = [ColumnConfig(**c) for c in ai_result.get("columns", [])]
+        result = orch.fill_table(
+            table_name=table_name,
+            count=count,
+            column_configs=column_configs,
+        )
+
+        return {
+            "model": ai_config.model,
+            "backend": ai_config.backend.value,
+            "table_name": result.table_name,
+            "count": result.count,
+            "elapsed": result.elapsed,
+            "errors": result.errors,
+            "ai_config": ai_result,
+        }
+
+
+@mcp.tool()
+def sqlseed_list_gemma_models() -> dict[str, Any]:
+    """List available Gemma 4 model variants with descriptions.
+
+    Returns information about all supported Gemma 4 models,
+    including recommended use cases for each variant.
+    """
+    models = []
+    for member in GemmaModel:
+        models.append({
+            "id": member.value,
+            "display_name": member.display_name,
+        })
+
+    backends = [
+        {"id": "google_ai_studio", "description": "Google AI Studio API (free tier available, recommended)"},
+        {"id": "lm_studio", "description": "LM Studio local deployment (http://127.0.0.1:1234, GUI-based)"},
+        {"id": "ollama", "description": "Ollama local deployment (offline, CLI-based)"},
+        {"id": "openai_compat", "description": "Any OpenAI-compatible API endpoint"},
+    ]
+
+    return {
+        "models": models,
+        "backends": backends,
+        "default_model": "gemma-4-26b-it",
+        "default_backend": "google_ai_studio",
+    }

plugins/sqlseed-ai/pyproject.toml

@@ -22,8 +22,9 @@ classifiers = [
     "Programming Language :: Python :: 3.13",
 ]
 dependencies = [
-    "sqlseed>=0.1.0,<2",
+    "sqlseed>=0.0.1",
     "openai>=1.0",
+    "google-generativeai>=0.8",
 ]
 
 [project.urls]

plugins/sqlseed-ai/src/sqlseed_ai/_client.py

@@ -14,11 +14,22 @@ def get_openai_client(config: Any | None = None) -> Any:
     if config is None:
         config = AIConfig.from_env()
 
+    # Use the new to_openai_kwargs() for unified resolution
+    if isinstance(config, AIConfig):
+        kwargs = config.to_openai_kwargs()
+        logger.info("Creating OpenAI client", **{"backend": config.backend.value, "base_url": kwargs["base_url"]})
+        return OpenAI(**kwargs)
+
+    # Legacy fallback for non-AIConfig objects
     api_key = config.api_key if hasattr(config, "api_key") else None
     base_url = config.base_url if hasattr(config, "base_url") else None
     timeout = config.timeout if hasattr(config, "timeout") else 60.0
 
     if not api_key:
-        raise ValueError("AI API key not configured. Set SQLSEED_AI_API_KEY or OPENAI_API_KEY environment variable.")
+        raise ValueError(
+            "AI API key not configured. "
+            "Set GOOGLE_API_KEY, SQLSEED_AI_API_KEY, or OPENAI_API_KEY environment variable. "
+            "For Ollama, set SQLSEED_AI_BACKEND=ollama."
+        )
 
     return OpenAI(api_key=api_key, base_url=base_url, timeout=timeout)

plugins/sqlseed-ai/src/sqlseed_ai/_model_selector.py

@@ -1,111 +1,98 @@
 from __future__ import annotations
 
-import json
-import time
-import urllib.error
-import urllib.request
-from typing import Any
+from sqlseed_ai.config import AIBackend, GemmaModel
 
 from sqlseed._utils.logger import get_logger
 
 logger = get_logger(__name__)
 
-_CACHE: dict[str, Any] = {
-    "model": None,
-    "expires_at": 0.0,
-    "available_models": [],
+# ── Gemma 4 model selection priority ────────────────────────────────
+# Ordered by capability: 26B MoE (best balance) > 31B Dense > 4B > 2B
+_GEMMA_MODEL_PRIORITY: list[GemmaModel] = [
+    GemmaModel.GEMMA_4_26B,
+    GemmaModel.GEMMA_4_31B,
+    GemmaModel.GEMMA_4_4B,
+    GemmaModel.GEMMA_4_2B,
+]
+
+# Map backend to preferred model size
+_BACKEND_DEFAULT_MODEL: dict[AIBackend, GemmaModel] = {
+    AIBackend.GOOGLE_AI_STUDIO: GemmaModel.GEMMA_4_26B,
+    AIBackend.LM_STUDIO: GemmaModel.GEMMA_4_4B,  # local inference, prefer smaller
+    AIBackend.OLLAMA: GemmaModel.GEMMA_4_4B,  # smaller for local inference
+    AIBackend.OPENAI_COMPAT: GemmaModel.GEMMA_4_26B,
 }
 
-_CACHE_TTL = 3600
 
-_OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models"
+def select_gemma_model(
+    backend: AIBackend = AIBackend.GOOGLE_AI_STUDIO,
+    prefer_small: bool = False,
+) -> str:
+    """Select the best Gemma 4 model for the given backend.
 
+    Args:
+        backend: The LLM backend provider.
+        prefer_small: If True, prefer smaller models (useful for Edge/local).
 
-def _fetch_available_free_models() -> list[str]:
-    try:
-        req = urllib.request.Request(_OPENROUTER_MODELS_URL)
-        with urllib.request.urlopen(req, timeout=10) as resp:
-            data = json.loads(resp.read().decode())
-    except (OSError, json.JSONDecodeError) as e:
-        logger.warning("Failed to fetch OpenRouter models, using fallback", error=str(e))
-        return []
+    Returns:
+        The model identifier string.
+    """
+    if prefer_small or backend in (AIBackend.OLLAMA, AIBackend.LM_STUDIO):
+        # For local inference (Ollama/LM Studio), prefer smaller models
+        model = GemmaModel.GEMMA_4_4B
+        logger.info("Selected compact Gemma 4 model for local inference", model=model.value)
+        return model.value
 
-    models_info = []
-    for model in data.get("data", []):
-        pricing = model.get("pricing", {})
-        if pricing.get("prompt") != "0" or pricing.get("completion") != "0":
-            continue
+    model = _BACKEND_DEFAULT_MODEL.get(backend, GemmaModel.GEMMA_4_26B)
+    logger.info("Selected Gemma 4 model", model=model.value, backend=backend.value)
+    return model.value
 
-        if model.get("expiration_date") is not None:
-            continue
 
-        arch = model.get("architecture", {})
-        if "text" not in arch.get("input_modalities", []):
-            continue
-        if "text" not in arch.get("output_modalities", []):
-            continue
+def select_next_gemma_model(failed_model: str) -> str | None:
+    """Select the next smaller Gemma 4 model as fallback.
 
-        supported = model.get("supported_parameters", [])
-        if "response_format" not in supported:
-            continue
+    Args:
+        failed_model: The model that failed.
 
-        models_info.append({"id": model["id"], "created": model.get("created", 0)})
+    Returns:
+        The next model in the priority list, or None if all exhausted.
+    """
+    for i, m in enumerate(_GEMMA_MODEL_PRIORITY):
+        if m.value == failed_model and i + 1 < len(_GEMMA_MODEL_PRIORITY):
+            next_model = _GEMMA_MODEL_PRIORITY[i + 1]
+            logger.info(
+                "Falling back to smaller Gemma 4 model",
+                from_model=failed_model,
+                to_model=next_model.value,
+            )
+            return next_model.value
 
-    models_info.sort(key=lambda x: x["created"], reverse=True)
-    return [m["id"] for m in models_info]
+    logger.warning("No more Gemma 4 models available for fallback", failed_model=failed_model)
+    return None
 
 
-def _update_cache(model: str) -> None:
-    _CACHE["model"] = model
-    _CACHE["expires_at"] = time.time() + _CACHE_TTL
+def get_available_gemma_models() -> list[dict[str, str]]:
+    """Return list of available Gemma 4 models with display info."""
+    return [
+        {"id": m.value, "display_name": m.display_name}
+        for m in _GEMMA_MODEL_PRIORITY
+    ]
 
 
-def select_best_free_model() -> str:
-    if _CACHE["model"] is not None and time.time() < _CACHE["expires_at"]:
-        return str(_CACHE["model"])
-
-    available = _fetch_available_free_models()
-    _CACHE["available_models"] = available
+# ── Legacy compatibility ─────────────────────────────────────────────
+# These functions maintain backward compatibility with code that
+# referenced the old OpenRouter-based model selector.
 
-    if available:
-        best = available[0]
-        _update_cache(best)
-        logger.info(
-            "Auto-selected newest free model from OpenRouter",
-            model=best,
-            available_count=len(available),
-        )
-        return best
-
-    fallback = "openrouter/free"
-    logger.warning("No free models without expiration could be fetched, using hardcoded fallback", model=fallback)
-    _update_cache(fallback)
-    logger.info("Using fallback free model", model=fallback)
-    return fallback
+def select_best_free_model() -> str:
+    """Legacy compat: returns the default Gemma 4 model."""
+    return select_gemma_model()
 
 
 def select_next_free_model(failed_model: str) -> str | None:
-    available: list[str] = _CACHE.get("available_models", [])
-    if not available:
-        available = _fetch_available_free_models()
-        _CACHE["available_models"] = available
-
-    idx = -1
-    for i, m in enumerate(available):
-        if m == failed_model:
-            idx = i
-            break
-
-    if idx == -1 or idx + 1 >= len(available):
-        return None
-
-    next_model = available[idx + 1]
-    _update_cache(next_model)
-    logger.info("Falling back to next free model", from_model=failed_model, to_model=next_model)
-    return next_model
+    """Legacy compat: returns the next Gemma 4 model as fallback."""
+    return select_next_gemma_model(failed_model)
 
 
 def clear_cache() -> None:
-    _CACHE["model"] = None
-    _CACHE["expires_at"] = 0.0
-    _CACHE["available_models"] = []
+    """Legacy compat: no-op, Gemma models don't need cache."""
+    pass