Gemini 2.5 and 3.0 MCP Client Setup Guide

sequenceDiagram
    participant User as Developer / Antigravity IDE
    participant Gemini as Gemini 2.5 / 3.0 Model
    participant Bridge as Gemini MCP Bridge
    participant Server as MCP Server (PostgreSQL)

    User->>Bridge: User prompt ("Find active users in db")
    Bridge->>Server: tools/list (JSON-RPC)
    Server-->>Bridge: Tool schemas (query, inspect_table)
    Bridge->>Gemini: Prompt + FunctionDeclarations
    Gemini-->>Bridge: functionCall: query(sql="SELECT * FROM users...")
    Bridge->>Server: tools/call (name="query", args={...})
    Server-->>Bridge: result: { rows: [...] }
    Bridge->>Gemini: functionResponse: { rows: [...] }
    Gemini-->>User: Formatted analytical response

{
  "mcpServers": {
    "postgres": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-postgres",
        "postgresql://admin:secret@/app_development"
      ]
    },
    "github": {
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_xxxxxxxxxxxxxxxxxxxx"
      }
    },
    "filesystem": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-filesystem",
        "c:/Users/developer/projects"
      ]
    }
  }
}

# gemini_mcp_client.py
import asyncio
import json
from google import genai
from google.genai import types

class GeminiMCPClient:
    def __init__(self, api_key: str, mcp_cmd: str, mcp_args: list[str]):
        self.ai = genai.Client(api_key=api_key)
        self.cmd = mcp_cmd
        self.args = mcp_args
        self.server = None

    async def start(self):
        """Spawn the MCP server process over stdio."""
        self.server = await asyncio.create_subprocess_exec(
            self.cmd, *self.args,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE
        )
        # Initialize MCP handshake
        await self._rpc_call("initialize", {
            "protocolVersion": "2024-11-05",
            "capabilities": {},
            "clientInfo": {"name": "Gemini-MCP-Runner", "version": "1.0"}
        })

    async def get_gemini_tool_declarations(self):
        """Fetch MCP tools and convert to Gemini FunctionDeclaration format."""
        tools_res = await self._rpc_call("tools/list", {})
        mcp_tools = tools_res.get("tools", [])

        function_declarations = []
        for tool in mcp_tools:
            function_declarations.append(
                types.FunctionDeclaration(
                    name=tool["name"],
                    description=tool.get("description", ""),
                    parameters=tool.get("inputSchema", {})
                )
            )
        return types.Tool(function_declarations=function_declarations)

    async def execute_tool_call(self, name: str, args: dict):
        """Relay Gemini tool call back into MCP tools/call."""
        res = await self._rpc_call("tools/call", {
            "name": name,
            "arguments": args
        })
        return res.get("content", [])

    async def _rpc_call(self, method: str, params: dict):
        payload = {"jsonrpc": "2.0", "id": 1, "method": method, "params": params}
        self.server.stdin.write((json.dumps(payload) + "\n").encode())
        await self.server.stdin.drain()
        line = await self.server.stdout.readline()
        return json.loads(line.decode()).get("result", {})