git-show · July 12, 2025 02:21
diff --git a/README.md b/README.md
diff --git a/gemini-config.json b/gemini-config.json
 {
  "enabled": true,
  "auto_consult": true,
  "cli_command": "gemini",
  "timeout": 30,
  "rate_limit_delay": 5.0,
  "log_consultations": true,
  "model": "gemini-2.5-pro",
  "sandbox_mode": false,
  "debug_mode": false,
  "include_history": true,
  "max_history_entries": 10,
  "uncertainty_thresholds": {
    "uncertainty_patterns": true,
    "complex_decisions": true,
    "critical_operations": true
  }
 }
diff --git a/gemini_integration.py b/gemini_integration.py
 #!/usr/bin/env python3
 """
 Gemini CLI Integration Module MCP Server
 Provides automatic consultation with Gemini for second opinions and validation
 """

 import asyncio
 import json
 import logging
 import re
 import subprocess
 import time
 from datetime import datetime
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple

 # Setup logging
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger(__name__)

 # Uncertainty patterns that trigger automatic Gemini consultation
 UNCERTAINTY_PATTERNS = [
    r"\bI'm not sure\b", r"\bI think\b", r"\bpossibly\b", r"\bprobably\b",
    r"\bmight be\b", r"\bcould be\b", r"\bI believe\b", r"\bIt seems\b",
    r"\bappears to be\b", r"\buncertain\b", r"\bI would guess\b",
    r"\blikely\b", r"\bperhaps\b", r"\bmaybe\b", r"\bI assume\b"
 ]

 # Complex decision patterns that benefit from second opinions
 COMPLEX_DECISION_PATTERNS = [
    r"\bmultiple approaches\b", r"\bseveral options\b", r"\btrade-offs?\b",
    r"\bconsider(?:ing)?\b", r"\balternatives?\b", r"\bpros and cons\b",
    r"\bweigh(?:ing)? the options\b", r"\bchoice between\b", r"\bdecision\b"
 ]

 # Critical operations that should trigger consultation
 CRITICAL_OPERATION_PATTERNS = [
    r"\bproduction\b", r"\bdatabase migration\b", r"\bsecurity\b",
    r"\bauthentication\b", r"\bencryption\b", r"\bAPI key\b",
    r"\bcredentials?\b", r"\bperformance\s+critical\b"
 ]


 class GeminiIntegration:
    """Handles Gemini CLI integration for second opinions and validation"""
    
    def __init__(self, config: Optional[Dict[str, Any]] = None):
        self.config = config or {}
        self.enabled = self.config.get('enabled', True)
        self.auto_consult = self.config.get('auto_consult', True)
        self.cli_command = self.config.get('cli_command', 'gemini')
        self.timeout = self.config.get('timeout', 60)
        self.rate_limit_delay = self.config.get('rate_limit_delay', 2.0)
        self.last_consultation = 0
        self.consultation_log = []
        self.max_context_length = self.config.get('max_context_length', 4000)
        self.model = self.config.get('model', 'gemini-2.5-flash')
        
        # Conversation history for maintaining state
        self.conversation_history = []
        self.max_history_entries = self.config.get('max_history_entries', 10)
        self.include_history = self.config.get('include_history', True)

    async def consult_gemini(self, query: str, context: str = "", 
                           comparison_mode: bool = True, 
                           force_consult: bool = False) -> Dict[str, Any]:
        """Consult Gemini CLI for second opinion"""
        if not self.enabled and not force_consult:
            return {'status': 'disabled', 'message': 'Gemini integration is disabled'}
        
        if not force_consult:
            await self._enforce_rate_limit()
        
        consultation_id = f"consult_{int(time.time())}_{len(self.consultation_log)}"
        
        try:
            # Prepare query with context
            full_query = self._prepare_query(query, context, comparison_mode)
            
            # Execute Gemini CLI command
            result = await self._execute_gemini_cli(full_query)
            
            # Save to conversation history
            if self.include_history and result.get('output'):
                self.conversation_history.append((query, result['output']))
                # Trim history if it exceeds max entries
                if len(self.conversation_history) > self.max_history_entries:
                    self.conversation_history = self.conversation_history[-self.max_history_entries:]
            
            # Log consultation
            if self.config.get('log_consultations', True):
                self.consultation_log.append({
                    'id': consultation_id,
                    'timestamp': datetime.now().isoformat(),
                    'query': query[:200] + "..." if len(query) > 200 else query,
                    'status': 'success',
                    'execution_time': result.get('execution_time', 0)
                })
            
            return {
                'status': 'success',
                'response': result['output'],
                'execution_time': result['execution_time'],
                'consultation_id': consultation_id,
                'timestamp': datetime.now().isoformat()
            }
            
        except Exception as e:
            logger.error(f"Error consulting Gemini: {str(e)}")
            return {
                'status': 'error',
                'error': str(e),
                'consultation_id': consultation_id
            }

    def detect_uncertainty(self, text: str) -> Tuple[bool, List[str]]:
        """Detect if text contains uncertainty patterns"""
        found_patterns = []
        
        # Check uncertainty patterns
        for pattern in UNCERTAINTY_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                found_patterns.append(f"uncertainty: {pattern}")
        
        # Check complex decision patterns
        for pattern in COMPLEX_DECISION_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                found_patterns.append(f"complex_decision: {pattern}")
        
        # Check critical operation patterns
        for pattern in CRITICAL_OPERATION_PATTERNS:
            if re.search(pattern, text, re.IGNORECASE):
                found_patterns.append(f"critical_operation: {pattern}")
        
        return len(found_patterns) > 0, found_patterns

    def clear_conversation_history(self) -> Dict[str, Any]:
        """Clear the conversation history"""
        old_count = len(self.conversation_history)
        self.conversation_history = []
        return {
            'status': 'success',
            'cleared_entries': old_count,
            'message': f'Cleared {old_count} conversation entries'
        }

    def get_consultation_stats(self) -> Dict[str, Any]:
        """Get statistics about consultations"""
        if not self.consultation_log:
            return {'total_consultations': 0}
            
        completed = [e for e in self.consultation_log if e.get('status') == 'success']
        
        return {
            'total_consultations': len(self.consultation_log),
            'completed_consultations': len(completed),
            'average_execution_time': sum(e.get('execution_time', 0) for e in completed) / len(completed) if completed else 0,
            'conversation_history_size': len(self.conversation_history)
        }

    async def _enforce_rate_limit(self):
        """Enforce rate limiting between consultations"""
        current_time = time.time()
        time_since_last = current_time - self.last_consultation
        
        if time_since_last < self.rate_limit_delay:
            sleep_time = self.rate_limit_delay - time_since_last
            await asyncio.sleep(sleep_time)
        
        self.last_consultation = time.time()

    def _prepare_query(self, query: str, context: str, comparison_mode: bool) -> str:
        """Prepare the full query for Gemini CLI"""
        parts = []
        
        if comparison_mode:
            parts.append("Please provide a technical analysis and second opinion:")
            parts.append("")
        
        # Include conversation history if enabled and available
        if self.include_history and self.conversation_history:
            parts.append("Previous conversation:")
            parts.append("-" * 40)
            for i, (prev_q, prev_a) in enumerate(self.conversation_history[-self.max_history_entries:], 1):
                parts.append(f"Q{i}: {prev_q}")
                # Truncate long responses in history
                if len(prev_a) > 500:
                    parts.append(f"A{i}: {prev_a[:500]}... [truncated]")
                else:
                    parts.append(f"A{i}: {prev_a}")
                parts.append("")
            parts.append("-" * 40)
            parts.append("")
        
        # Truncate context if too long
        if len(context) > self.max_context_length:
            context = context[:self.max_context_length] + "\n[Context truncated...]"
        
        if context:
            parts.append("Context:")
            parts.append(context)
            parts.append("")
        
        parts.append("Current Question/Topic:")
        parts.append(query)
        
        if comparison_mode:
            parts.extend([
                "",
                "Please structure your response with:",
                "1. Your analysis and understanding",
                "2. Recommendations or approach", 
                "3. Any concerns or considerations",
                "4. Alternative approaches (if applicable)"
            ])
        
        return "\n".join(parts)

    async def _execute_gemini_cli(self, query: str) -> Dict[str, Any]:
        """Execute Gemini CLI command and return results"""
        start_time = time.time()
        
        # Build command
        cmd = [self.cli_command]
        if self.model:
            cmd.extend(['-m', self.model])
        cmd.extend(['-p', query])  # Non-interactive mode
        
        try:
            process = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            
            stdout, stderr = await asyncio.wait_for(
                process.communicate(),
                timeout=self.timeout
            )
            
            execution_time = time.time() - start_time
            
            if process.returncode != 0:
                error_msg = stderr.decode() if stderr else "Unknown error"
                if "authentication" in error_msg.lower():
                    error_msg += "\nTip: Run 'gemini' interactively to authenticate"
                raise Exception(f"Gemini CLI failed: {error_msg}")
            
            return {
                'output': stdout.decode().strip(),
                'execution_time': execution_time
            }
            
        except asyncio.TimeoutError:
            raise Exception(f"Gemini CLI timed out after {self.timeout} seconds")


 # Singleton pattern implementation
 _integration = None


 def get_integration(config: Optional[Dict[str, Any]] = None) -> GeminiIntegration:
    """
    Get or create the global Gemini integration instance.
    
    This ensures that all parts of the application share the same instance,
    maintaining consistent state for rate limiting, consultation history,
    and configuration across all tool calls.
    
    Args:
        config: Optional configuration dict. Only used on first call.
    
    Returns:
        The singleton GeminiIntegration instance
    """
    global _integration
    if _integration is None:
        _integration = GeminiIntegration(config)
    return _integration
diff --git a/mcp-server.py b/mcp-server.py
 #!/usr/bin/env python3
 """
 MCP Server with Gemini Integration
 Provides development workflow automation with AI second opinions
 """

 import asyncio
 import json
 import os
 import sys
 from pathlib import Path
 from typing import Any, Dict, List, Optional, Tuple

 import mcp.server.stdio
 import mcp.types as types
 from mcp.server import NotificationOptions, Server, InitializationOptions

 # Assuming gemini_integration.py is in the same directory or properly installed
 from gemini_integration import get_integration


 class MCPServer:
    def __init__(self, project_root: str = None):
        self.project_root = Path(project_root) if project_root else Path.cwd()
        self.server = Server("gemini-mcp-server")
        
        # Initialize Gemini integration with singleton pattern
        self.gemini_config = self._load_gemini_config()
        # Get the singleton instance, passing config on first call
        self.gemini = get_integration(self.gemini_config)
        
        # Track uncertainty for auto-consultation
        self.last_response_uncertainty = None
        
        self._setup_tools()

    def _load_gemini_config(self) -> Dict[str, Any]:
        """Load Gemini configuration from environment or config file."""
        # Try to load .env file if it exists
        env_file = self.project_root / '.env'
        if env_file.exists():
            try:
                with open(env_file, 'r') as f:
                    for line in f:
                        line = line.strip()
                        if line and not line.startswith('#') and '=' in line:
                            key, value = line.split('=', 1)
                            # Only set if not already in environment
                            if key not in os.environ:
                                os.environ[key] = value
            except Exception as e:
                print(f"Warning: Could not load .env file: {e}")
        
        config = {
            'enabled': os.getenv('GEMINI_ENABLED', 'true').lower() == 'true',
            'auto_consult': os.getenv('GEMINI_AUTO_CONSULT', 'true').lower() == 'true',
            'cli_command': os.getenv('GEMINI_CLI_COMMAND', 'gemini'),
            'timeout': int(os.getenv('GEMINI_TIMEOUT', '60')),
            'rate_limit_delay': float(os.getenv('GEMINI_RATE_LIMIT', '2')),
            'max_context_length': int(os.getenv('GEMINI_MAX_CONTEXT', '4000')),
            'log_consultations': os.getenv('GEMINI_LOG_CONSULTATIONS', 'true').lower() == 'true',
            'model': os.getenv('GEMINI_MODEL', 'gemini-2.5-flash'),
            'sandbox_mode': os.getenv('GEMINI_SANDBOX', 'false').lower() == 'true',
            'debug_mode': os.getenv('GEMINI_DEBUG', 'false').lower() == 'true',
            'include_history': os.getenv('GEMINI_INCLUDE_HISTORY', 'true').lower() == 'true',
            'max_history_entries': int(os.getenv('GEMINI_MAX_HISTORY', '10')),
        }
        
        # Try to load from config file
        config_file = self.project_root / 'gemini-config.json'
        if config_file.exists():
            try:
                with open(config_file, 'r') as f:
                    file_config = json.load(f)
                config.update(file_config)
            except Exception as e:
                print(f"Warning: Could not load gemini-config.json: {e}")
        
        return config

    def _setup_tools(self):
        """Register all MCP tools"""
        
        # Gemini consultation tool
        @self.server.call_tool()
        async def consult_gemini(arguments: Dict[str, Any]) -> List[types.TextContent]:
            """Consult Gemini CLI for a second opinion or validation."""
            query = arguments.get('query', '')
            context = arguments.get('context', '')
            comparison_mode = arguments.get('comparison_mode', True)
            force_consult = arguments.get('force', False)
            
            if not query:
                return [types.TextContent(
                    type="text",
                    text="❌ Error: 'query' parameter is required for Gemini consultation"
                )]
            
            # Consult Gemini
            result = await self.gemini.consult_gemini(
                query=query,
                context=context,
                comparison_mode=comparison_mode,
                force_consult=force_consult
            )
            
            # Format the response
            return await self._format_gemini_response(result)
        
        @self.server.call_tool()
        async def gemini_status(arguments: Dict[str, Any]) -> List[types.TextContent]:
            """Get Gemini integration status and statistics."""
            return await self._get_gemini_status()
        
        @self.server.call_tool()
        async def toggle_gemini_auto_consult(arguments: Dict[str, Any]) -> List[types.TextContent]:
            """Toggle automatic Gemini consultation on uncertainty detection."""
            enable = arguments.get('enable', None)
            
            if enable is None:
                # Toggle current state
                self.gemini.auto_consult = not self.gemini.auto_consult
            else:
                self.gemini.auto_consult = bool(enable)
            
            status = "enabled" if self.gemini.auto_consult else "disabled"
            return [types.TextContent(
                type="text",
                text=f"✅ Gemini auto-consultation is now {status}"
            )]
        
        @self.server.call_tool()
        async def clear_gemini_history(arguments: Dict[str, Any]) -> List[types.TextContent]:
            """Clear Gemini conversation history."""
            result = self.gemini.clear_conversation_history()
            return [types.TextContent(
                type="text",
                text=f"✅ {result['message']}"
            )]

    async def _format_gemini_response(self, result: Dict[str, Any]) -> List[types.TextContent]:
        """Format Gemini consultation response for MCP output."""
        output_lines = []
        output_lines.append("🤖 Gemini Consultation Response")
        output_lines.append("=" * 40)
        output_lines.append("")
        
        if result['status'] == 'success':
            output_lines.append(f"✅ Consultation ID: {result['consultation_id']}")
            output_lines.append(f"⏱️  Execution time: {result['execution_time']:.2f}s")
            output_lines.append("")
            
            # Display the raw response (simplified format)
            response = result.get('response', '')
            if response:
                output_lines.append("📄 Response:")
                output_lines.append(response)
        
        elif result['status'] == 'disabled':
            output_lines.append("ℹ️  Gemini consultation is currently disabled")
            output_lines.append("💡 Enable with: toggle_gemini_auto_consult")
        
        elif result['status'] == 'timeout':
            output_lines.append(f"❌ {result['error']}")
            output_lines.append("💡 Try increasing the timeout or simplifying the query")
        
        else:  # error
            output_lines.append(f"❌ Error: {result.get('error', 'Unknown error')}")
            output_lines.append("")
            output_lines.append("💡 Troubleshooting:")
            output_lines.append("  1. Check if Gemini CLI is installed and in PATH")
            output_lines.append("  2. Verify Gemini CLI authentication")
            output_lines.append("  3. Check the logs for more details")
        
        return [types.TextContent(type="text", text="\n".join(output_lines))]
    
    async def _get_gemini_status(self) -> List[types.TextContent]:
        """Get Gemini integration status and statistics."""
        output_lines = []
        output_lines.append("🤖 Gemini Integration Status")
        output_lines.append("=" * 40)
        output_lines.append("")
        
        # Configuration status
        output_lines.append("⚙️  Configuration:")
        output_lines.append(f"  • Enabled: {'✅ Yes' if self.gemini.enabled else '❌ No'}")
        output_lines.append(f"  • Auto-consult: {'✅ Yes' if self.gemini.auto_consult else '❌ No'}")
        output_lines.append(f"  • CLI command: {self.gemini.cli_command}")
        output_lines.append(f"  • Timeout: {self.gemini.timeout}s")
        output_lines.append(f"  • Rate limit: {self.gemini.rate_limit_delay}s")
        output_lines.append("")
        
        # Check if Gemini CLI is available
        try:
            # Test with a simple prompt rather than --version (which may not be supported)
            check_process = await asyncio.create_subprocess_exec(
                self.gemini.cli_command, "-p", "test",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE
            )
            stdout, stderr = await asyncio.wait_for(check_process.communicate(), timeout=10)
            
            if check_process.returncode == 0:
                output_lines.append("✅ Gemini CLI is available and working")
                # Try to get version info from help or other means
                try:
                    help_process = await asyncio.create_subprocess_exec(
                        self.gemini.cli_command, "--help",
                        stdout=asyncio.subprocess.PIPE,
                        stderr=asyncio.subprocess.PIPE
                    )
                    help_stdout, _ = await help_process.communicate()
                    help_text = help_stdout.decode()
                    # Look for version in help output
                    if "version" in help_text.lower():
                        for line in help_text.split('\n'):
                            if 'version' in line.lower():
                                output_lines.append(f"  {line.strip()}")
                                break
                except:
                    pass
            else:
                error_msg = stderr.decode() if stderr else "Unknown error"
                output_lines.append("❌ Gemini CLI found but not working properly")
                output_lines.append(f"  Command tested: {self.gemini.cli_command}")
                output_lines.append(f"  Error: {error_msg}")
                
                # Check for authentication issues
                if "authentication" in error_msg.lower() or "api key" in error_msg.lower():
                    output_lines.append("")
                    output_lines.append("🔑 Authentication required:")
                    output_lines.append("  1. Set GEMINI_API_KEY environment variable, or")
                    output_lines.append("  2. Run 'gemini' interactively to authenticate with Google")
                    
        except asyncio.TimeoutError:
            output_lines.append("❌ Gemini CLI test timed out")
            output_lines.append("  This may indicate authentication is required")
        except FileNotFoundError:
            output_lines.append("❌ Gemini CLI not found in PATH")
            output_lines.append(f"  Expected command: {self.gemini.cli_command}")
            output_lines.append("")
            output_lines.append("📦 Installation:")
            output_lines.append("  npm install -g @google/gemini-cli")
            output_lines.append("  OR")
            output_lines.append("  npx @google/gemini-cli")
        except Exception as e:
            output_lines.append(f"❌ Error checking Gemini CLI: {str(e)}")
        
        output_lines.append("")
        
        # Consultation statistics
        stats = self.gemini.get_consultation_stats()
        output_lines.append("📊 Consultation Statistics:")
        output_lines.append(f"  • Total consultations: {stats.get('total_consultations', 0)}")
        
        completed = stats.get('completed_consultations', 0)
        output_lines.append(f"  • Completed: {completed}")
        
        if completed > 0:
            avg_time = stats.get('average_execution_time', 0)
            total_time = stats.get('total_execution_time', 0)
            output_lines.append(f"  • Average time: {avg_time:.2f}s")
            output_lines.append(f"  • Total time: {total_time:.2f}s")
        
        last_consultation = stats.get('last_consultation')
        if last_consultation:
            output_lines.append(f"  • Last consultation: {last_consultation}")
        
        output_lines.append("")
        output_lines.append("💡 Usage:")
        output_lines.append("  • Direct: Use 'consult_gemini' tool")
        output_lines.append("  • Auto: Enable auto-consult for uncertainty detection")
        output_lines.append("  • Toggle: Use 'toggle_gemini_auto_consult' tool")
        
        return [types.TextContent(type="text", text="\n".join(output_lines))]
    
    def detect_response_uncertainty(self, response: str) -> Tuple[bool, List[str]]:
        """
        Detect uncertainty in a response for potential auto-consultation.
        This is a wrapper around the GeminiIntegration's detection.
        """
        return self.gemini.detect_uncertainty(response)
    
    async def maybe_consult_gemini(self, response: str, context: str = "") -> Optional[Dict[str, Any]]:
        """
        Check if response contains uncertainty and consult Gemini if needed.
        
        Args:
            response: The response to check for uncertainty
            context: Additional context for the consultation
            
        Returns:
            Gemini consultation result if consulted, None otherwise
        """
        if not self.gemini.auto_consult or not self.gemini.enabled:
            return None
        
        has_uncertainty, patterns = self.detect_response_uncertainty(response)
        
        if has_uncertainty:
            # Extract the main question or topic from the response
            query = f"Please provide a second opinion on this analysis:\n\n{response}"
            
            # Add uncertainty patterns to context
            enhanced_context = f"{context}\n\nUncertainty detected in: {', '.join(patterns)}"
            
            result = await self.gemini.consult_gemini(
                query=query,
                context=enhanced_context,
                comparison_mode=True
            )
            
            return result
        
        return None
    
    def run(self):
        """Run the MCP server."""
        async def main():
            async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
                await self.server.run(
                    read_stream,
                    write_stream,
                    InitializationOptions(
                        server_name="gemini-mcp-server",
                        server_version="1.0.0",
                        capabilities=self.server.get_capabilities(
                            notification_options=NotificationOptions(),
                            experimental_capabilities={},
                        ),
                    ),
                )
        
        asyncio.run(main())


 if __name__ == "__main__":
    import argparse
    
    parser = argparse.ArgumentParser(description="MCP Server with Gemini Integration")
    parser.add_argument(
        "--project-root",
        type=str,
        default=".",
        help="Path to the project root directory"
    )
    
    args = parser.parse_args()
    
    server = MCPServer(args.project_root)
    server.run()
diff --git a/setup-gemini-integration.sh b/setup-gemini-integration.sh
 #!/bin/bash
 set -e

 echo "🚀 Setting up Gemini CLI Integration..."

 # Check Node.js version
 if ! command -v node &> /dev/null; then
    echo "❌ Node.js not found. Please install Node.js 18+ first."
    exit 1
 fi

 NODE_VERSION=$(node --version | cut -d'v' -f2 | cut -d'.' -f1)
 if [ "$NODE_VERSION" -lt 18 ]; then
    echo "❌ Node.js version $NODE_VERSION found. Please use Node.js 18+ (recommended: 22.16.0)"
    echo "   Use: nvm install 22.16.0 && nvm use 22.16.0"
    exit 1
 fi

 echo "✅ Node.js version check passed"

 # Install Gemini CLI
 echo "📦 Installing Gemini CLI..."
 npm install -g @google/gemini-cli

 # Test installation
 echo "🧪 Testing Gemini CLI installation..."
 if gemini --help > /dev/null 2>&1; then
    echo "✅ Gemini CLI installed successfully"
 else
    echo "❌ Gemini CLI installation failed"
    exit 1
 fi

 # Files can be placed in the same directory - no complex structure needed
 echo "📁 Setting up in current directory..."

 # Create default configuration
 echo "⚙️ Creating default configuration..."
 cat > gemini-config.json << 'EOF'
 {
  "enabled": true,
  "auto_consult": true,
  "cli_command": "gemini",
  "timeout": 60,
  "rate_limit_delay": 2.0,
  "max_context_length": 4000,
  "log_consultations": true,
  "model": "gemini-2.5-flash",
  "sandbox_mode": false,
  "debug_mode": false
 }
 EOF

 # Create MCP configuration for Claude Code
 echo "🔧 Creating Claude Code MCP configuration..."
 cat > mcp-config.json << 'EOF'
 {
  "mcpServers": {
    "project": {
      "command": "python3",
      "args": ["mcp-server.py", "--project-root", "."],
      "env": {
        "GEMINI_ENABLED": "true",
        "GEMINI_AUTO_CONSULT": "true"
      }
    }
  }
 }
 EOF

 echo ""
 echo "🎉 Gemini CLI Integration setup complete!"
 echo ""
 echo "📋 Next steps:"
 echo "1. Copy the provided code files to your project:"
 echo "   - gemini_integration.py"
 echo "   - mcp-server.py"
 echo "2. Install Python dependencies: pip install mcp pydantic"
 echo "3. Test with: python3 mcp-server.py --project-root ."
 echo "4. Configure Claude Code to use the MCP server"
 echo ""
 echo "💡 Tip: First run 'gemini' command to authenticate with your Google account"
diff --git a/test_gemini_state.py b/test_gemini_state.py
 #!/usr/bin/env python3
 """Test Gemini state management with automatic history inclusion."""

 import asyncio
 import sys
 from pathlib import Path

 from gemini_integration import get_integration

 async def test_automatic_state():
    """Test if Gemini automatically maintains state through history."""
    print("🧪 Testing Automatic Gemini State Management")
    print("=" * 50)
    
    # Use singleton
    gemini = get_integration({
        'enabled': True,
        'cli_command': 'gemini',
        'timeout': 30,
        'include_history': True,  # This enables automatic history inclusion
        'max_history_entries': 10,
        'debug_mode': False
    })
    
    # Clear any existing history
    gemini.clear_conversation_history()
    
    print("\n1️⃣  First Question: What is 2+2?")
    print("-" * 30)
    
    try:
        response1 = await gemini.consult_gemini(
            query="What is 2+2?",
            context="",  # No context needed
            force_consult=True
        )
        
        if response1.get('status') == 'success':
            response_text = response1.get('response', '')
            print(f"✅ Success! Gemini responded.")
            
            # Find and show the answer
            lines = response_text.strip().split('\n')
            for line in lines:
                if '4' in line or 'four' in line.lower():
                    print(f"📝 Found answer: {line.strip()[:100]}...")
                    break
        else:
            print(f"❌ Error: {response1.get('error', 'Unknown error')}")
            return
            
    except Exception as e:
        print(f"❌ Exception: {e}")
        return
    
    print(f"\n📊 Conversation history size: {len(gemini.conversation_history)}")
    
    print("\n2️⃣  Second Question: What is that doubled?")
    print("   (No context provided - relying on automatic history)")
    print("-" * 30)
    
    try:
        # This time, provide NO context at all - let the history do the work
        response2 = await gemini.consult_gemini(
            query="What is that doubled?",
            context="",  # Empty context - history should provide the context
            force_consult=True
        )
        
        if response2.get('status') == 'success':
            response_text = response2.get('response', '')
            print(f"✅ Success! Gemini responded.")
            
            # Check if it understood the context
            if '8' in response_text or 'eight' in response_text.lower():
                print("🎉 STATE MAINTAINED! Gemini understood 'that' referred to 4")
                print("📝 Found reference to 8 in the response")
                
                # Find and show where 8 appears
                for line in response_text.split('\n'):
                    if '8' in line or 'eight' in line.lower():
                        print(f"📝 Context: {line.strip()[:100]}...")
                        break
            else:
                print("⚠️  Gemini may not have maintained state properly")
                print("📝 Response doesn't clearly reference 8")
                print(f"First 200 chars: {response_text[:200]}...")
                
        else:
            print(f"❌ Error: {response2.get('error', 'Unknown error')}")
            
    except Exception as e:
        print(f"❌ Exception: {e}")
    
    print(f"\n📊 Final conversation history size: {len(gemini.conversation_history)}")
    
    # Test with history disabled
    print("\n3️⃣  Testing with History Disabled")
    print("-" * 30)
    
    # Disable history
    gemini.include_history = False
    gemini.clear_conversation_history()
    
    # Ask first question
    await gemini.consult_gemini(
        query="What is 3+3?",
        context="",
        force_consult=True
    )
    
    # Ask follow-up
    response3 = await gemini.consult_gemini(
        query="What is that tripled?",
        context="",
        force_consult=True
    )
    
    if response3.get('status') == 'success':
        response_text = response3.get('response', '')
        if '18' in response_text or 'eighteen' in response_text.lower():
            print("❌ UNEXPECTED: Found 18 even without history!")
        else:
            print("✅ EXPECTED: Without history, Gemini doesn't understand 'that'")
            # Show what Gemini says when it doesn't have context
            print(f"Response preview: {response_text[:200]}...")
    
    print("\n✅ Test complete!")

 if __name__ == "__main__":
    asyncio.run(test_automatic_state())
Issue	Solution
Gemini CLI not found	Install Node.js 18+ and `npm install -g @google/gemini-cli`
Authentication errors	Run `gemini` and sign in with Google account
Node version issues	Use `nvm use 22.16.0`
Timeout errors	Increase `GEMINI_TIMEOUT` (default: 60s)
Auto-consult not working	Check `GEMINI_AUTO_CONSULT=true`
Rate limiting	Adjust `GEMINI_RATE_LIMIT` (default: 2s)
	{
	"enabled": true,
	"auto_consult": true,
	"cli_command": "gemini",
	"timeout": 30,
	"rate_limit_delay": 5.0,
	"log_consultations": true,
	"model": "gemini-2.5-pro",
	"sandbox_mode": false,
	"debug_mode": false,
	"include_history": true,
	"max_history_entries": 10,
	"uncertainty_thresholds": {
	"uncertainty_patterns": true,
	"complex_decisions": true,
	"critical_operations": true
	}
	}
	#!/usr/bin/env python3
	"""
	Gemini CLI Integration Module MCP Server
	Provides automatic consultation with Gemini for second opinions and validation
	"""

	import asyncio
	import json
	import logging
	import re
	import subprocess
	import time
	from datetime import datetime
	from pathlib import Path
	from typing import Any, Dict, List, Optional, Tuple

	# Setup logging
	logging.basicConfig(level=logging.INFO)
	logger = logging.getLogger(__name__)

	# Uncertainty patterns that trigger automatic Gemini consultation
	UNCERTAINTY_PATTERNS = [
	r"\bI'm not sure\b", r"\bI think\b", r"\bpossibly\b", r"\bprobably\b",
	r"\bmight be\b", r"\bcould be\b", r"\bI believe\b", r"\bIt seems\b",
	r"\bappears to be\b", r"\buncertain\b", r"\bI would guess\b",
	r"\blikely\b", r"\bperhaps\b", r"\bmaybe\b", r"\bI assume\b"
	]

	# Complex decision patterns that benefit from second opinions
	COMPLEX_DECISION_PATTERNS = [
	r"\bmultiple approaches\b", r"\bseveral options\b", r"\btrade-offs?\b",
	r"\bconsider(?:ing)?\b", r"\balternatives?\b", r"\bpros and cons\b",
	r"\bweigh(?:ing)? the options\b", r"\bchoice between\b", r"\bdecision\b"
	]

	# Critical operations that should trigger consultation
	CRITICAL_OPERATION_PATTERNS = [
	r"\bproduction\b", r"\bdatabase migration\b", r"\bsecurity\b",
	r"\bauthentication\b", r"\bencryption\b", r"\bAPI key\b",
	r"\bcredentials?\b", r"\bperformance\s+critical\b"
	]


	class GeminiIntegration:
	"""Handles Gemini CLI integration for second opinions and validation"""

	def __init__(self, config: Optional[Dict[str, Any]] = None):
	self.config = config or {}
	self.enabled = self.config.get('enabled', True)
	self.auto_consult = self.config.get('auto_consult', True)
	self.cli_command = self.config.get('cli_command', 'gemini')
	self.timeout = self.config.get('timeout', 60)
	self.rate_limit_delay = self.config.get('rate_limit_delay', 2.0)
	self.last_consultation = 0
	self.consultation_log = []
	self.max_context_length = self.config.get('max_context_length', 4000)
	self.model = self.config.get('model', 'gemini-2.5-flash')

	# Conversation history for maintaining state
	self.conversation_history = []
	self.max_history_entries = self.config.get('max_history_entries', 10)
	self.include_history = self.config.get('include_history', True)

	async def consult_gemini(self, query: str, context: str = "",
	comparison_mode: bool = True,
	force_consult: bool = False) -> Dict[str, Any]:
	"""Consult Gemini CLI for second opinion"""
	if not self.enabled and not force_consult:
	return {'status': 'disabled', 'message': 'Gemini integration is disabled'}

	if not force_consult:
	await self._enforce_rate_limit()

	consultation_id = f"consult_{int(time.time())}_{len(self.consultation_log)}"

	try:
	# Prepare query with context
	full_query = self._prepare_query(query, context, comparison_mode)

	# Execute Gemini CLI command
	result = await self._execute_gemini_cli(full_query)

	# Save to conversation history
	if self.include_history and result.get('output'):
	self.conversation_history.append((query, result['output']))
	# Trim history if it exceeds max entries
	if len(self.conversation_history) > self.max_history_entries:
	self.conversation_history = self.conversation_history[-self.max_history_entries:]

	# Log consultation
	if self.config.get('log_consultations', True):
	self.consultation_log.append({
	'id': consultation_id,
	'timestamp': datetime.now().isoformat(),
	'query': query[:200] + "..." if len(query) > 200 else query,
	'status': 'success',
	'execution_time': result.get('execution_time', 0)
	})

	return {
	'status': 'success',
	'response': result['output'],
	'execution_time': result['execution_time'],
	'consultation_id': consultation_id,
	'timestamp': datetime.now().isoformat()
	}

	except Exception as e:
	logger.error(f"Error consulting Gemini: {str(e)}")
	return {
	'status': 'error',
	'error': str(e),
	'consultation_id': consultation_id
	}

	def detect_uncertainty(self, text: str) -> Tuple[bool, List[str]]:
	"""Detect if text contains uncertainty patterns"""
	found_patterns = []

	# Check uncertainty patterns
	for pattern in UNCERTAINTY_PATTERNS:
	if re.search(pattern, text, re.IGNORECASE):
	found_patterns.append(f"uncertainty: {pattern}")

	# Check complex decision patterns
	for pattern in COMPLEX_DECISION_PATTERNS:
	if re.search(pattern, text, re.IGNORECASE):
	found_patterns.append(f"complex_decision: {pattern}")

	# Check critical operation patterns
	for pattern in CRITICAL_OPERATION_PATTERNS:
	if re.search(pattern, text, re.IGNORECASE):
	found_patterns.append(f"critical_operation: {pattern}")

	return len(found_patterns) > 0, found_patterns

	def clear_conversation_history(self) -> Dict[str, Any]:
	"""Clear the conversation history"""
	old_count = len(self.conversation_history)
	self.conversation_history = []
	return {
	'status': 'success',
	'cleared_entries': old_count,
	'message': f'Cleared {old_count} conversation entries'
	}

	def get_consultation_stats(self) -> Dict[str, Any]:
	"""Get statistics about consultations"""
	if not self.consultation_log:
	return {'total_consultations': 0}

	completed = [e for e in self.consultation_log if e.get('status') == 'success']

	return {
	'total_consultations': len(self.consultation_log),
	'completed_consultations': len(completed),
	'average_execution_time': sum(e.get('execution_time', 0) for e in completed) / len(completed) if completed else 0,
	'conversation_history_size': len(self.conversation_history)
	}

	async def _enforce_rate_limit(self):
	"""Enforce rate limiting between consultations"""
	current_time = time.time()
	time_since_last = current_time - self.last_consultation

	if time_since_last < self.rate_limit_delay:
	sleep_time = self.rate_limit_delay - time_since_last
	await asyncio.sleep(sleep_time)

	self.last_consultation = time.time()

	def _prepare_query(self, query: str, context: str, comparison_mode: bool) -> str:
	"""Prepare the full query for Gemini CLI"""
	parts = []

	if comparison_mode:
	parts.append("Please provide a technical analysis and second opinion:")
	parts.append("")

	# Include conversation history if enabled and available
	if self.include_history and self.conversation_history:
	parts.append("Previous conversation:")
	parts.append("-" * 40)
	for i, (prev_q, prev_a) in enumerate(self.conversation_history[-self.max_history_entries:], 1):
	parts.append(f"Q{i}: {prev_q}")
	# Truncate long responses in history
	if len(prev_a) > 500:
	parts.append(f"A{i}: {prev_a[:500]}... [truncated]")
	else:
	parts.append(f"A{i}: {prev_a}")
	parts.append("")
	parts.append("-" * 40)
	parts.append("")

	# Truncate context if too long
	if len(context) > self.max_context_length:
	context = context[:self.max_context_length] + "\n[Context truncated...]"

	if context:
	parts.append("Context:")
	parts.append(context)
	parts.append("")

	parts.append("Current Question/Topic:")
	parts.append(query)

	if comparison_mode:
	parts.extend([
	"",
	"Please structure your response with:",
	"1. Your analysis and understanding",
	"2. Recommendations or approach",
	"3. Any concerns or considerations",
	"4. Alternative approaches (if applicable)"
	])

	return "\n".join(parts)

	async def _execute_gemini_cli(self, query: str) -> Dict[str, Any]:
	"""Execute Gemini CLI command and return results"""
	start_time = time.time()

	# Build command
	cmd = [self.cli_command]
	if self.model:
	cmd.extend(['-m', self.model])
	cmd.extend(['-p', query]) # Non-interactive mode

	try:
	process = await asyncio.create_subprocess_exec(
	*cmd,
	stdout=asyncio.subprocess.PIPE,
	stderr=asyncio.subprocess.PIPE
	)

	stdout, stderr = await asyncio.wait_for(
	process.communicate(),
	timeout=self.timeout
	)

	execution_time = time.time() - start_time

	if process.returncode != 0:
	error_msg = stderr.decode() if stderr else "Unknown error"
	if "authentication" in error_msg.lower():
	error_msg += "\nTip: Run 'gemini' interactively to authenticate"
	raise Exception(f"Gemini CLI failed: {error_msg}")

	return {
	'output': stdout.decode().strip(),
	'execution_time': execution_time
	}

	except asyncio.TimeoutError:
	raise Exception(f"Gemini CLI timed out after {self.timeout} seconds")


	# Singleton pattern implementation
	_integration = None


	def get_integration(config: Optional[Dict[str, Any]] = None) -> GeminiIntegration:
	"""
	Get or create the global Gemini integration instance.

	This ensures that all parts of the application share the same instance,
	maintaining consistent state for rate limiting, consultation history,
	and configuration across all tool calls.

	Args:
	config: Optional configuration dict. Only used on first call.

	Returns:
	The singleton GeminiIntegration instance
	"""
	global _integration
	if _integration is None:
	_integration = GeminiIntegration(config)
	return _integration
	#!/usr/bin/env python3
	"""
	MCP Server with Gemini Integration
	Provides development workflow automation with AI second opinions
	"""

	import asyncio
	import json
	import os
	import sys
	from pathlib import Path
	from typing import Any, Dict, List, Optional, Tuple

	import mcp.server.stdio
	import mcp.types as types
	from mcp.server import NotificationOptions, Server, InitializationOptions

	# Assuming gemini_integration.py is in the same directory or properly installed
	from gemini_integration import get_integration


	class MCPServer:
	def __init__(self, project_root: str = None):
	self.project_root = Path(project_root) if project_root else Path.cwd()
	self.server = Server("gemini-mcp-server")

	# Initialize Gemini integration with singleton pattern
	self.gemini_config = self._load_gemini_config()
	# Get the singleton instance, passing config on first call
	self.gemini = get_integration(self.gemini_config)

	# Track uncertainty for auto-consultation
	self.last_response_uncertainty = None

	self._setup_tools()

	def _load_gemini_config(self) -> Dict[str, Any]:
	"""Load Gemini configuration from environment or config file."""
	# Try to load .env file if it exists
	env_file = self.project_root / '.env'
	if env_file.exists():
	try:
	with open(env_file, 'r') as f:
	for line in f:
	line = line.strip()
	if line and not line.startswith('#') and '=' in line:
	key, value = line.split('=', 1)
	# Only set if not already in environment
	if key not in os.environ:
	os.environ[key] = value
	except Exception as e:
	print(f"Warning: Could not load .env file: {e}")

	config = {
	'enabled': os.getenv('GEMINI_ENABLED', 'true').lower() == 'true',
	'auto_consult': os.getenv('GEMINI_AUTO_CONSULT', 'true').lower() == 'true',
	'cli_command': os.getenv('GEMINI_CLI_COMMAND', 'gemini'),
	'timeout': int(os.getenv('GEMINI_TIMEOUT', '60')),
	'rate_limit_delay': float(os.getenv('GEMINI_RATE_LIMIT', '2')),
	'max_context_length': int(os.getenv('GEMINI_MAX_CONTEXT', '4000')),
	'log_consultations': os.getenv('GEMINI_LOG_CONSULTATIONS', 'true').lower() == 'true',
	'model': os.getenv('GEMINI_MODEL', 'gemini-2.5-flash'),
	'sandbox_mode': os.getenv('GEMINI_SANDBOX', 'false').lower() == 'true',
	'debug_mode': os.getenv('GEMINI_DEBUG', 'false').lower() == 'true',
	'include_history': os.getenv('GEMINI_INCLUDE_HISTORY', 'true').lower() == 'true',
	'max_history_entries': int(os.getenv('GEMINI_MAX_HISTORY', '10')),
	}

	# Try to load from config file
	config_file = self.project_root / 'gemini-config.json'
	if config_file.exists():
	try:
	with open(config_file, 'r') as f:
	file_config = json.load(f)
	config.update(file_config)
	except Exception as e:
	print(f"Warning: Could not load gemini-config.json: {e}")

	return config

	def _setup_tools(self):
	"""Register all MCP tools"""

	# Gemini consultation tool
	@self.server.call_tool()
	async def consult_gemini(arguments: Dict[str, Any]) -> List[types.TextContent]:
	"""Consult Gemini CLI for a second opinion or validation."""
	query = arguments.get('query', '')
	context = arguments.get('context', '')
	comparison_mode = arguments.get('comparison_mode', True)
	force_consult = arguments.get('force', False)

	if not query:
	return [types.TextContent(
	type="text",
	text="❌ Error: 'query' parameter is required for Gemini consultation"
	)]

	# Consult Gemini
	result = await self.gemini.consult_gemini(
	query=query,
	context=context,
	comparison_mode=comparison_mode,
	force_consult=force_consult
	)

	# Format the response
	return await self._format_gemini_response(result)

	@self.server.call_tool()
	async def gemini_status(arguments: Dict[str, Any]) -> List[types.TextContent]:
	"""Get Gemini integration status and statistics."""
	return await self._get_gemini_status()

	@self.server.call_tool()
	async def toggle_gemini_auto_consult(arguments: Dict[str, Any]) -> List[types.TextContent]:
	"""Toggle automatic Gemini consultation on uncertainty detection."""
	enable = arguments.get('enable', None)

	if enable is None:
	# Toggle current state
	self.gemini.auto_consult = not self.gemini.auto_consult
	else:
	self.gemini.auto_consult = bool(enable)

	status = "enabled" if self.gemini.auto_consult else "disabled"
	return [types.TextContent(
	type="text",
	text=f"✅ Gemini auto-consultation is now {status}"
	)]

	@self.server.call_tool()
	async def clear_gemini_history(arguments: Dict[str, Any]) -> List[types.TextContent]:
	"""Clear Gemini conversation history."""
	result = self.gemini.clear_conversation_history()
	return [types.TextContent(
	type="text",
	text=f"✅ {result['message']}"
	)]

	async def _format_gemini_response(self, result: Dict[str, Any]) -> List[types.TextContent]:
	"""Format Gemini consultation response for MCP output."""
	output_lines = []
	output_lines.append("🤖 Gemini Consultation Response")
	output_lines.append("=" * 40)
	output_lines.append("")

	if result['status'] == 'success':
	output_lines.append(f"✅ Consultation ID: {result['consultation_id']}")
	output_lines.append(f"⏱️ Execution time: {result['execution_time']:.2f}s")
	output_lines.append("")

	# Display the raw response (simplified format)
	response = result.get('response', '')
	if response:
	output_lines.append("📄 Response:")
	output_lines.append(response)

	elif result['status'] == 'disabled':
	output_lines.append("ℹ️ Gemini consultation is currently disabled")
	output_lines.append("💡 Enable with: toggle_gemini_auto_consult")

	elif result['status'] == 'timeout':
	output_lines.append(f"❌ {result['error']}")
	output_lines.append("💡 Try increasing the timeout or simplifying the query")

	else: # error
	output_lines.append(f"❌ Error: {result.get('error', 'Unknown error')}")
	output_lines.append("")
	output_lines.append("💡 Troubleshooting:")
	output_lines.append(" 1. Check if Gemini CLI is installed and in PATH")
	output_lines.append(" 2. Verify Gemini CLI authentication")
	output_lines.append(" 3. Check the logs for more details")

	return [types.TextContent(type="text", text="\n".join(output_lines))]

	async def _get_gemini_status(self) -> List[types.TextContent]:
	"""Get Gemini integration status and statistics."""
	output_lines = []
	output_lines.append("🤖 Gemini Integration Status")
	output_lines.append("=" * 40)
	output_lines.append("")

	# Configuration status
	output_lines.append("⚙️ Configuration:")
	output_lines.append(f" • Enabled: {'✅ Yes' if self.gemini.enabled else '❌ No'}")
	output_lines.append(f" • Auto-consult: {'✅ Yes' if self.gemini.auto_consult else '❌ No'}")
	output_lines.append(f" • CLI command: {self.gemini.cli_command}")
	output_lines.append(f" • Timeout: {self.gemini.timeout}s")
	output_lines.append(f" • Rate limit: {self.gemini.rate_limit_delay}s")
	output_lines.append("")

	# Check if Gemini CLI is available
	try:
	# Test with a simple prompt rather than --version (which may not be supported)
	check_process = await asyncio.create_subprocess_exec(
	self.gemini.cli_command, "-p", "test",
	stdout=asyncio.subprocess.PIPE,
	stderr=asyncio.subprocess.PIPE
	)
	stdout, stderr = await asyncio.wait_for(check_process.communicate(), timeout=10)

	if check_process.returncode == 0:
	output_lines.append("✅ Gemini CLI is available and working")
	# Try to get version info from help or other means
	try:
	help_process = await asyncio.create_subprocess_exec(
	self.gemini.cli_command, "--help",
	stdout=asyncio.subprocess.PIPE,
	stderr=asyncio.subprocess.PIPE
	)
	help_stdout, _ = await help_process.communicate()
	help_text = help_stdout.decode()
	# Look for version in help output
	if "version" in help_text.lower():
	for line in help_text.split('\n'):
	if 'version' in line.lower():
	output_lines.append(f" {line.strip()}")
	break
	except:
	pass
	else:
	error_msg = stderr.decode() if stderr else "Unknown error"
	output_lines.append("❌ Gemini CLI found but not working properly")
	output_lines.append(f" Command tested: {self.gemini.cli_command}")
	output_lines.append(f" Error: {error_msg}")

	# Check for authentication issues
	if "authentication" in error_msg.lower() or "api key" in error_msg.lower():
	output_lines.append("")
	output_lines.append("🔑 Authentication required:")
	output_lines.append(" 1. Set GEMINI_API_KEY environment variable, or")
	output_lines.append(" 2. Run 'gemini' interactively to authenticate with Google")

	except asyncio.TimeoutError:
	output_lines.append("❌ Gemini CLI test timed out")
	output_lines.append(" This may indicate authentication is required")
	except FileNotFoundError:
	output_lines.append("❌ Gemini CLI not found in PATH")
	output_lines.append(f" Expected command: {self.gemini.cli_command}")
	output_lines.append("")
	output_lines.append("📦 Installation:")
	output_lines.append(" npm install -g @google/gemini-cli")
	output_lines.append(" OR")
	output_lines.append(" npx @google/gemini-cli")
	except Exception as e:
	output_lines.append(f"❌ Error checking Gemini CLI: {str(e)}")

	output_lines.append("")

	# Consultation statistics
	stats = self.gemini.get_consultation_stats()
	output_lines.append("📊 Consultation Statistics:")
	output_lines.append(f" • Total consultations: {stats.get('total_consultations', 0)}")

	completed = stats.get('completed_consultations', 0)
	output_lines.append(f" • Completed: {completed}")

	if completed > 0:
	avg_time = stats.get('average_execution_time', 0)
	total_time = stats.get('total_execution_time', 0)
	output_lines.append(f" • Average time: {avg_time:.2f}s")
	output_lines.append(f" • Total time: {total_time:.2f}s")

	last_consultation = stats.get('last_consultation')
	if last_consultation:
	output_lines.append(f" • Last consultation: {last_consultation}")

	output_lines.append("")
	output_lines.append("💡 Usage:")
	output_lines.append(" • Direct: Use 'consult_gemini' tool")
	output_lines.append(" • Auto: Enable auto-consult for uncertainty detection")
	output_lines.append(" • Toggle: Use 'toggle_gemini_auto_consult' tool")

	return [types.TextContent(type="text", text="\n".join(output_lines))]

	def detect_response_uncertainty(self, response: str) -> Tuple[bool, List[str]]:
	"""
	Detect uncertainty in a response for potential auto-consultation.
	This is a wrapper around the GeminiIntegration's detection.
	"""
	return self.gemini.detect_uncertainty(response)

	async def maybe_consult_gemini(self, response: str, context: str = "") -> Optional[Dict[str, Any]]:
	"""
	Check if response contains uncertainty and consult Gemini if needed.

	Args:
	response: The response to check for uncertainty
	context: Additional context for the consultation

	Returns:
	Gemini consultation result if consulted, None otherwise
	"""
	if not self.gemini.auto_consult or not self.gemini.enabled:
	return None

	has_uncertainty, patterns = self.detect_response_uncertainty(response)

	if has_uncertainty:
	# Extract the main question or topic from the response
	query = f"Please provide a second opinion on this analysis:\n\n{response}"

	# Add uncertainty patterns to context
	enhanced_context = f"{context}\n\nUncertainty detected in: {', '.join(patterns)}"

	result = await self.gemini.consult_gemini(
	query=query,
	context=enhanced_context,
	comparison_mode=True
	)

	return result

	return None

	def run(self):
	"""Run the MCP server."""
	async def main():
	async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
	await self.server.run(
	read_stream,
	write_stream,
	InitializationOptions(
	server_name="gemini-mcp-server",
	server_version="1.0.0",
	capabilities=self.server.get_capabilities(
	notification_options=NotificationOptions(),
	experimental_capabilities={},
	),
	),
	)

	asyncio.run(main())


	if __name__ == "__main__":
	import argparse

	parser = argparse.ArgumentParser(description="MCP Server with Gemini Integration")
	parser.add_argument(
	"--project-root",
	type=str,
	default=".",
	help="Path to the project root directory"
	)

	args = parser.parse_args()

	server = MCPServer(args.project_root)
	server.run()
	#!/bin/bash
	set -e

	echo "🚀 Setting up Gemini CLI Integration..."

	# Check Node.js version
	if ! command -v node &> /dev/null; then
	echo "❌ Node.js not found. Please install Node.js 18+ first."
	exit 1
	fi

	NODE_VERSION=$(node --version \| cut -d'v' -f2 \| cut -d'.' -f1)
	if [ "$NODE_VERSION" -lt 18 ]; then
	echo "❌ Node.js version $NODE_VERSION found. Please use Node.js 18+ (recommended: 22.16.0)"
	echo " Use: nvm install 22.16.0 && nvm use 22.16.0"
	exit 1
	fi

	echo "✅ Node.js version check passed"

	# Install Gemini CLI
	echo "📦 Installing Gemini CLI..."
	npm install -g @google/gemini-cli

	# Test installation
	echo "🧪 Testing Gemini CLI installation..."
	if gemini --help > /dev/null 2>&1; then
	echo "✅ Gemini CLI installed successfully"
	else
	echo "❌ Gemini CLI installation failed"
	exit 1
	fi

	# Files can be placed in the same directory - no complex structure needed
	echo "📁 Setting up in current directory..."

	# Create default configuration
	echo "⚙️ Creating default configuration..."
	cat > gemini-config.json << 'EOF'
	{
	"enabled": true,
	"auto_consult": true,
	"cli_command": "gemini",
	"timeout": 60,
	"rate_limit_delay": 2.0,
	"max_context_length": 4000,
	"log_consultations": true,
	"model": "gemini-2.5-flash",
	"sandbox_mode": false,
	"debug_mode": false
	}
	EOF

	# Create MCP configuration for Claude Code
	echo "🔧 Creating Claude Code MCP configuration..."
	cat > mcp-config.json << 'EOF'
	{
	"mcpServers": {
	"project": {
	"command": "python3",
	"args": ["mcp-server.py", "--project-root", "."],
	"env": {
	"GEMINI_ENABLED": "true",
	"GEMINI_AUTO_CONSULT": "true"
	}
	}
	}
	}
	EOF

	echo ""
	echo "🎉 Gemini CLI Integration setup complete!"
	echo ""
	echo "📋 Next steps:"
	echo "1. Copy the provided code files to your project:"
	echo " - gemini_integration.py"
	echo " - mcp-server.py"
	echo "2. Install Python dependencies: pip install mcp pydantic"
	echo "3. Test with: python3 mcp-server.py --project-root ."
	echo "4. Configure Claude Code to use the MCP server"
	echo ""
	echo "💡 Tip: First run 'gemini' command to authenticate with your Google account"
	#!/usr/bin/env python3
	"""Test Gemini state management with automatic history inclusion."""

	import asyncio
	import sys
	from pathlib import Path

	from gemini_integration import get_integration

	async def test_automatic_state():
	"""Test if Gemini automatically maintains state through history."""
	print("🧪 Testing Automatic Gemini State Management")
	print("=" * 50)

	# Use singleton
	gemini = get_integration({
	'enabled': True,
	'cli_command': 'gemini',
	'timeout': 30,
	'include_history': True, # This enables automatic history inclusion
	'max_history_entries': 10,
	'debug_mode': False
	})

	# Clear any existing history
	gemini.clear_conversation_history()

	print("\n1️⃣ First Question: What is 2+2?")
	print("-" * 30)

	try:
	response1 = await gemini.consult_gemini(
	query="What is 2+2?",
	context="", # No context needed
	force_consult=True
	)

	if response1.get('status') == 'success':
	response_text = response1.get('response', '')
	print(f"✅ Success! Gemini responded.")

	# Find and show the answer
	lines = response_text.strip().split('\n')
	for line in lines:
	if '4' in line or 'four' in line.lower():
	print(f"📝 Found answer: {line.strip()[:100]}...")
	break
	else:
	print(f"❌ Error: {response1.get('error', 'Unknown error')}")
	return

	except Exception as e:
	print(f"❌ Exception: {e}")
	return

	print(f"\n📊 Conversation history size: {len(gemini.conversation_history)}")

	print("\n2️⃣ Second Question: What is that doubled?")
	print(" (No context provided - relying on automatic history)")
	print("-" * 30)

	try:
	# This time, provide NO context at all - let the history do the work
	response2 = await gemini.consult_gemini(
	query="What is that doubled?",
	context="", # Empty context - history should provide the context
	force_consult=True
	)

	if response2.get('status') == 'success':
	response_text = response2.get('response', '')
	print(f"✅ Success! Gemini responded.")

	# Check if it understood the context
	if '8' in response_text or 'eight' in response_text.lower():
	print("🎉 STATE MAINTAINED! Gemini understood 'that' referred to 4")
	print("📝 Found reference to 8 in the response")

	# Find and show where 8 appears
	for line in response_text.split('\n'):
	if '8' in line or 'eight' in line.lower():
	print(f"📝 Context: {line.strip()[:100]}...")
	break
	else:
	print("⚠️ Gemini may not have maintained state properly")
	print("📝 Response doesn't clearly reference 8")
	print(f"First 200 chars: {response_text[:200]}...")

	else:
	print(f"❌ Error: {response2.get('error', 'Unknown error')}")

	except Exception as e:
	print(f"❌ Exception: {e}")

	print(f"\n📊 Final conversation history size: {len(gemini.conversation_history)}")

	# Test with history disabled
	print("\n3️⃣ Testing with History Disabled")
	print("-" * 30)

	# Disable history
	gemini.include_history = False
	gemini.clear_conversation_history()

	# Ask first question
	await gemini.consult_gemini(
	query="What is 3+3?",
	context="",
	force_consult=True
	)

	# Ask follow-up
	response3 = await gemini.consult_gemini(
	query="What is that tripled?",
	context="",
	force_consult=True
	)

	if response3.get('status') == 'success':
	response_text = response3.get('response', '')
	if '18' in response_text or 'eighteen' in response_text.lower():
	print("❌ UNEXPECTED: Found 18 even without history!")
	else:
	print("✅ EXPECTED: Without history, Gemini doesn't understand 'that'")
	# Show what Gemini says when it doesn't have context
	print(f"Response preview: {response_text[:200]}...")

	print("\n✅ Test complete!")

	if __name__ == "__main__":
	asyncio.run(test_automatic_state())