#!/usr/bin/env python3 """ HTTP-based Gemini MCP Server Provides REST API interface for Gemini consultation (for testing/development) """ import os import sys from pathlib import Path from typing import Any, Dict, Optional from fastapi import FastAPI, HTTPException from pydantic import BaseModel import uvicorn # Check if running in container BEFORE any other imports or operations def check_container_and_exit(): """Check if running in a container and exit immediately if true.""" if os.path.exists("/.dockerenv") or os.environ.get("CONTAINER_ENV"): print("ERROR: Gemini MCP Server cannot run inside a container!", file=sys.stderr) print( "The Gemini CLI requires Docker access and must run on the host system.", file=sys.stderr, ) print("Please launch this server directly on the host with:", file=sys.stderr) print(" python gemini_mcp_server_http.py", file=sys.stderr) sys.exit(1) # Perform container check immediately check_container_and_exit() # Now import the integration module from gemini_integration import get_integration # Initialize FastAPI app app = FastAPI( title="Gemini MCP Server (HTTP Mode)", version="1.0.0", description="HTTP interface for Gemini CLI integration" ) class ConsultRequest(BaseModel): query: str context: Optional[str] = "" comparison_mode: Optional[bool] = True force: Optional[bool] = False class ToggleRequest(BaseModel): enable: Optional[bool] = None class ConsultResponse(BaseModel): status: str response: Optional[str] = None error: Optional[str] = None consultation_id: Optional[str] = None execution_time: Optional[float] = None timestamp: Optional[str] = None # Initialize Gemini integration project_root = Path(os.environ.get("PROJECT_ROOT", ".")) 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'), } # Get the singleton instance gemini = get_integration(config) @app.get("/") async def root(): """Root endpoint showing server info""" return { "name": "Gemini MCP Server (HTTP Mode)", "version": "1.0.0", "mode": "http", "endpoints": { "health": "/health", "tools": "/mcp/tools", "consult": "/mcp/tools/consult_gemini", "status": "/mcp/tools/gemini_status", "toggle": "/mcp/tools/toggle_gemini_auto_consult", "clear_history": "/mcp/tools/clear_gemini_history" } } @app.get("/health") async def health(): """Health check endpoint""" return {"status": "healthy", "mode": "http"} @app.get("/mcp/tools") async def list_tools(): """List available MCP tools""" return { "tools": [ { "name": "consult_gemini", "description": "Consult Gemini CLI for a second opinion or validation", "parameters": { "query": {"type": "string", "required": True}, "context": {"type": "string", "required": False}, "comparison_mode": {"type": "boolean", "required": False, "default": True}, "force": {"type": "boolean", "required": False, "default": False} } }, { "name": "gemini_status", "description": "Get Gemini integration status and statistics", "parameters": {} }, { "name": "toggle_gemini_auto_consult", "description": "Toggle automatic Gemini consultation on uncertainty detection", "parameters": { "enable": {"type": "boolean", "required": False} } }, { "name": "clear_gemini_history", "description": "Clear Gemini conversation history", "parameters": {} } ] } @app.post("/mcp/tools/consult_gemini", response_model=ConsultResponse) async def consult_gemini_endpoint(request: ConsultRequest): """Consult Gemini for a second opinion""" if not request.query: raise HTTPException(status_code=400, detail="Query parameter is required") result = await gemini.consult_gemini( query=request.query, context=request.context, comparison_mode=request.comparison_mode, force_consult=request.force ) return ConsultResponse(**result) @app.get("/mcp/tools/gemini_status") async def gemini_status_endpoint(): """Get Gemini integration status and statistics""" import asyncio # Build status information status = { "configuration": { "enabled": gemini.enabled, "auto_consult": gemini.auto_consult, "cli_command": gemini.cli_command, "timeout": gemini.timeout, "rate_limit": gemini.rate_limit_delay, "model": gemini.model }, "statistics": gemini.get_consultation_stats() } # Check CLI availability try: check_process = await asyncio.create_subprocess_exec( gemini.cli_command, "-p", "test", stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) stdout, stderr = await asyncio.wait_for(check_process.communicate(), timeout=5) status["cli_available"] = check_process.returncode == 0 if not status["cli_available"]: status["cli_error"] = stderr.decode() if stderr else "Unknown error" except Exception as e: status["cli_available"] = False status["cli_error"] = str(e) return status @app.post("/mcp/tools/toggle_gemini_auto_consult") async def toggle_auto_consult_endpoint(request: ToggleRequest): """Toggle automatic Gemini consultation""" if request.enable is None: # Toggle current state gemini.auto_consult = not gemini.auto_consult else: gemini.auto_consult = bool(request.enable) return { "status": "success", "auto_consult": gemini.auto_consult, "message": f"Gemini auto-consultation is now {'enabled' if gemini.auto_consult else 'disabled'}" } @app.post("/mcp/tools/clear_gemini_history") async def clear_history_endpoint(): """Clear Gemini conversation history""" result = gemini.clear_conversation_history() return result def run_http_server(port: int): """Run the HTTP server""" host = os.environ.get("GEMINI_MCP_HOST", "127.0.0.1") print(f"Starting Gemini MCP Server in HTTP mode on {host}:{port}") print("WARNING: HTTP mode is for testing only. Use stdio mode for production.") print(f"Access the API at: http://{host}:{port}") print(f"Health check: http://{host}:{port}/health") print(f"API docs: http://{host}:{port}/docs") uvicorn.run(app, host=host, port=port) if __name__ == "__main__": # Check if running in container check_container_and_exit() # Default port from environment or 8006 default_port = int(os.environ.get("GEMINI_MCP_PORT", "8006")) # Run server run_http_server(default_port)