Feat/gemma3 model intelligence

examples/gemma3-mcp-agent/README.md

@@ -0,0 +1,63 @@
+# Gemma 3 & FunctionGemma MCP Gateway
+
+This repository provides an official implementation of a bridge between Google's **Gemma 3 / FunctionGemma** and the **Model Context Protocol (MCP)**.
+
+## 🚀 Installation
+
+Ensure you have a modern Python environment (3.10+) and run:
+
+```bash
+# Clone the repository and navigate to the example
+cd gemma/examples/gemma3-mcp-agent/
+
+# Install the required MCP and communication libraries
+pip install -r requirements.txt
+```
+
+### Prerequisites
+1. **Ollama**: Download from [ollama.com](https://ollama.com).
+2. **Gemma 3 Model**: Run `ollama pull gemma3`.
+
+---
+
+## 🏗️ The "Gemma 3 Bridge"
+
+This bridge is uniquely designed to bypass the common "regex parser" failures found in standard implementations. It utilizes the **Official Native Tokens** for high-reliability tool execution.
+
+### Key Features
+* **Model Intelligence Layer**: Automatically detects system RAM to recommend the optimal Gemma 3 size (1B, 4B, 12B, or 27B).
+* **Official Specification**: Aligns with `FunctionGemma` standards using the `declaration:tool_name{schema}` format.
+* **Native Transitions**: Uses official control tokens:
+    * `<start_function_call>` and `<end_function_call>`
+    * `<start_function_response>` and `<end_function_response>`
+* **Developer-Role Implementation**: Automatically injects the `developer` turn required to trigger Gemma 3's high-reasoning tool-use mode.
+* **Escape Handling**: Built-in support for the `<escape>` token, ensuring JSON inputs remain valid even with complex special characters.
+
+---
+
+## 🧪 Usage & Quick Start
+
+### 1. Using the MCP Inspector (Verification)
+To verify the bridge and inspect tool schemas without an IDE, use the `mcp-inspector`:
+
+```bash
+npx @modelcontextprotocol/inspector python server.py
+```
+* Once the inspector loads, you can view the `gemma_chat` tool.
+* You can trigger a test call to `get_system_info` or `read_local_file` to see the native token encapsulation in action.
+
+### 2. Integration with Antigravity IDE
+1. Open **Antigravity Settings**.
+2. Navigate to **MCP Servers**.
+3. Import the `mcp_config.json` provided in this directory.
+    * *Note: Ensure the `args` path in `mcp_config.json` correctly points to `server.py` relative to your workspace root.*
+4. The IDE agent will now be able to use Gemma 3 via the `gemma_chat` tool for local reasoning.
+
+### 3. Verification Test Case
+Ask the agent: 
+> "Check my system OS and read the content of requirements.txt."
+
+This will trigger a multi-turn reasoning loop:
+1. Model generates `<start_function_call>call:get_system_info{}<end_function_call>`.
+2. Gateway executes local check and returns `<start_function_response>`.
+3. Model generates the second call for `read_local_file`.

examples/gemma3-mcp-agent/config.py

@@ -0,0 +1,25 @@
+import os
+
+# Mapping of model size aliases to official Ollama/Gemma identifiers
+MODEL_MAP = {
+    "1b": "gemma3:1b",
+    "4b": "gemma3:4b",
+    "12b": "gemma3:12b",
+    "27b": "gemma3:27b",
+    "small": "gemma3:4b",
+    "medium": "gemma3:12b",
+    "large": "gemma3:27b"
+}
+
+def get_model_identifier():
+    """
+    Retrieves the model identifier based on environment variables or defaults.
+    Allows user override via GEMMA_MODEL_SIZE.
+    """
+    # Check for manual override
+    manual_size = os.getenv("GEMMA_MODEL_SIZE")
+    if manual_size and manual_size.lower() in MODEL_MAP:
+        return MODEL_MAP[manual_size.lower()]
+
+    # This will be supplemented by the hardware detector in system_utils.py
+    return None

examples/gemma3-mcp-agent/mcp_config.json

@@ -0,0 +1,14 @@
+{
+    "mcpServers": {
+        "gemma-gateway": {
+            "command": "python",
+            "args": [
+                "./examples/gemma3-mcp-agent/server.py"
+            ],
+            "env": {
+                "OLLAMA_URL": "http://localhost:11434/api/generate",
+                "GEMMA_MODEL": "gemma3"
+            }
+        }
+    }
+}

examples/gemma3-mcp-agent/requirements.txt

@@ -0,0 +1,5 @@
+mcp
+fastmcp
+httpx
+pydantic
+psutil

examples/gemma3-mcp-agent/server.py

@@ -0,0 +1,128 @@
+import os, json, re, httpx, asyncio
+from mcp.server.fastmcp import FastMCP
+
+# Import hardware intelligence layer
+from config import MODEL_MAP, get_model_identifier
+from system_utils import get_recommended_model, get_total_ram_gb
+
+mcp = FastMCP("Gemma-MCP-Gateway")
+
+# Configuration and Hardware Intelligence
+OLLAMA_URL = os.environ.get("OLLAMA_URL", "http://localhost:11434/api/generate")
+
+# Detect hardware and select model
+total_ram = get_total_ram_gb()
+recommended_size = get_recommended_model()
+default_model = MODEL_MAP[recommended_size]
+
+# Check for override in config or env
+env_model = get_model_identifier()
+MODEL_NAME = env_model if env_model else default_model
+
+print(f"--- Gemma MCP Gateway Startup ---")
+print(f"Detected {total_ram}GB RAM. Defaulting to {MODEL_NAME} for optimal performance.")
+print(f"---------------------------------")
+
+# Official trigger phrase for Gemma 3 function calling
+GEMMA_SYSTEM_PROMPT = "You are a model that can do function calling with the following functions"
+
+def format_tools_for_gemma(tools):
+    """Format tools using official declaration tokens."""
+    definitions = [f"declaration:{t.name}{json.dumps(t.input_schema)}" for t in tools]
+    return f"<start_function_declaration>\n" + "\n".join(definitions) + "\n<end_function_declaration>"
+
+def parse_gemma_tool_call(text):
+    """Parses official Gemma 3 tool calls, handling <escape> tokens."""
+    # Remove <escape> tokens if present before parsing JSON to prevent breakage
+    clean_text = text.replace("<escape>", "")
+
+    # Official native token pattern
+    call_regex = r"<start_function_call>call:(\w+)(\{.*?\})<end_function_call>"
+    match = re.search(call_regex, clean_text, re.DOTALL)
+
+    if match:
+        tool_name = match.group(1)
+        try:
+            return tool_name, json.loads(match.group(2)), ""
+        except:
+            return None, None, ""
+    return None, None, ""
+
+@mcp.tool()
+async def gemma_chat(prompt: str, history: list = None) -> str:
+    """
+    A tool-augmented chat interface utilizing official Gemma 3 'Native Token' strategies.
+    Handles developer role activation and recursive tool execution.
+    """
+    all_tools = mcp.list_tools()
+    available_tools = [t for t in all_tools if t.name != "gemma_chat"]
+
+    # Construct the Developer turn (Turn 1) - Official trigger for Tool-use Mode
+    tool_block = format_tools_for_gemma(available_tools)
+    full_prompt = f"<start_of_turn>developer\n{GEMMA_SYSTEM_PROMPT}{tool_block}<end_of_turn>\n"
+
+    # Append conversation history if provided
+    if history:
+        for turn in history:
+            full_prompt += f"<start_of_turn>{turn['role']}\n{turn['content']}<end_of_turn>\n"
+
+    # Add final user prompt
+    full_prompt += f"<start_of_turn>user\n{prompt}<end_of_turn>\n<start_of_turn>model\n"
+
+    async with httpx.AsyncClient() as client:
+        current_prompt = full_prompt
+        # Support up to 5 tool call rounds to prevent cycles
+        for _ in range(5):
+            response = await client.post(
+                OLLAMA_URL, 
+                json={
+                    "model": MODEL_NAME, 
+                    "prompt": current_prompt, 
+                    "stream": False, 
+                    "raw": True # Required for precise control token handling
+                }, 
+                timeout=120.0
+            )
+
+            if response.status_code != 200:
+                return f"Error from Ollama ({response.status_code}): {response.text}"
+
+            output = response.json().get("response", "")
+            tool_name, tool_args, _ = parse_gemma_tool_call(output)
+
+            if tool_name:
+                try:
+                    tool_result = await mcp.call_tool(tool_name, tool_args)
+
+                    # Official native response format
+                    res_block = f"<start_function_response>{tool_result}<end_function_response>"
+
+                    # Append result back to the prompt as a user turn continuation
+                    current_prompt += output + f"\n<start_of_turn>user\n{res_block}<end_of_turn>\n<start_of_turn>model\n"
+                except Exception as e:
+                    err_block = f"<start_function_response>Error: {str(e)}<end_function_response>"
+                    current_prompt += output + f"\n<start_of_turn>user\n{err_block}<end_of_turn>\n<start_of_turn>model\n"
+            else:
+                # No tool call detected, return final model output
+                return output
+
+    return "Max tool iterations reached."
+
+@mcp.tool()
+async def read_local_file(path: str) -> str:
+    """Reads a file from the local filesystem."""
+    try:
+        with open(path, 'r', encoding='utf-8') as f:
+            return f.read()
+    except Exception as e:
+        return f"Error reading file: {str(e)}"
+
+@mcp.tool()
+async def get_system_info() -> str:
+    """Get basic system information."""
+    import platform
+    return f"OS: {platform.system()} {platform.release()}, Arch: {platform.machine()}"
+
+if __name__ == "__main__":
+    # Single entry point for mcp server lifecycle
+    mcp.run()

examples/gemma3-mcp-agent/system_utils.py

@@ -0,0 +1,29 @@
+import psutil
+
+def get_total_ram_gb():
+    """
+    Detects the total system RAM and returns it in GB.
+    """
+    ram_bytes = psutil.virtual_memory().total
+    return round(ram_bytes / (1024 ** 3))
+
+def get_recommended_model():
+    """
+    Recommends a Gemma 3 model size based on detected system RAM.
+
+    Recommendation Logic:
+    - < 8GB: 1b
+    - 8GB - 16GB: 4b
+    - 16GB - 32GB: 12b
+    - > 32GB: 27b
+    """
+    ram_gb = get_total_ram_gb()
+
+    if ram_gb < 8:
+        return '1b'
+    elif 8 <= ram_gb < 16:
+        return '4b'
+    elif 16 <= ram_gb <= 32:
+        return '12b'
+    else:
+        return '27b'

examples/gemma3-mcp-agent/test_config.py

@@ -0,0 +1,30 @@
+import unittest
+import os
+from config import get_model_identifier, MODEL_MAP
+
+class TestConfig(unittest.TestCase):
+    """
+    Unit tests for the configuration and model selection logic.
+    """
+
+    def test_model_identifiers(self):
+        """Verify that all mapped aliases exist in the MODEL_MAP."""
+        self.assertIn("1b", MODEL_MAP)
+        self.assertIn("4b", MODEL_MAP)
+        self.assertIn("small", MODEL_MAP)
+        self.assertEqual(MODEL_MAP["small"], "gemma3:4b")
+
+    def test_environment_override(self):
+        """Verify that GEMMA_MODEL_SIZE environment variable correctly overrides the model."""
+        # Set environment variable
+        os.environ["GEMMA_MODEL_SIZE"] = "small"
+
+        # Test the detection
+        model = get_model_identifier()
+        self.assertEqual(model, "gemma3:4b")
+
+        # Clean up
+        del os.environ["GEMMA_MODEL_SIZE"]
+
+if __name__ == "__main__":
+    unittest.main()