"""
SSE Transport wrapper for Gemini CLI MCP Server (Fixed).
Includes fixes for stderr handling and environment validation.
"""
import os
import sys
from mcp.server.fastmcp import FastMCP
import subprocess
import re
import shutil
# Get configuration from environment
HOST = os.environ.get("MCP_SERVER_HOST", "0.0.0.0")
PORT = int(os.environ.get("MCP_SERVER_PORT", "8601"))
GEMINI_PATH = os.environ.get("GEMINI_PATH", "/home/ward/.nvm/versions/node/v20.19.5/bin/gemini")
# Create MCP server with settings via keyword args
mcp = FastMCP(
"Gemini CLI MCP Server",
host=HOST,
port=PORT,
)
def strip_ansi(text: str) -> str:
"""Remove ANSI escape sequences from string."""
ansi_escape = re.compile(r'\x1B(?:[@-Z\\-_]|[0-9a-zA-Z\[\]\{\}\(\)\%\\\/\~\^\`\|\<\>\!\@\#\$\&\*\=\+\?])')
return ansi_escape.sub('', text)
def filter_stderr(stderr: str, keep_extensions: bool = False) -> str:
"""
Filter out known noise from stderr.
Includes patterns for Gemini and generic Node/Copilot warnings.
"""
if not stderr:
return ""
noise_patterns = [
# Gemini Patterns
r'^\\[STARTUP\\]$.*$',
r'^Loading extension:.*$',
r'^Loaded cached credentials\.$',
r'^YOLO mode is enabled\..*$',
r'^\[ERROR\] \[ImportProcessor\].*parcel.*$',
# Generic Node/Copilot Patterns (Add your copilot specific noise here)
r'^.*node_modules.*$', # Filter stack trace paths in non-critical errors
r'^file://.*$', # Filter file URL references
r'^.*ExperimentalWarning.*$',
]
if not keep_extensions:
noise_patterns.extend([
r'^Installed extensions:$',
r'^- [a-z0-9-]+$',
])
compiled_patterns = [re.compile(p, re.MULTILINE) for p in noise_patterns]
lines = stderr.strip().split('\n')
filtered_lines = []
for line in lines:
line = line.strip()
if not line:
continue
is_noise = False
for pattern in compiled_patterns:
if pattern.match(line):
is_noise = True
break
if not is_noise:
filtered_lines.append(line)
return '\n'.join(filtered_lines)
def process_gemini_output(result: subprocess.CompletedProcess, keep_extensions: bool = False) -> str:
"""
Process output with strict stderr handling.
Only includes stderr if the command failed or stdout is empty.
"""
output = strip_ansi(result.stdout).strip() if result.stdout else ""
stderr_raw = strip_ansi(result.stderr).strip() if result.stderr else ""
filtered_stderr = filter_stderr(stderr_raw, keep_extensions)
# 1. Success case: return stdout
if result.returncode == 0:
if output:
return output
# If success but no stdout, show stderr (might be a warning or status)
return filtered_stderr if filtered_stderr else "Command succeeded (no output)"
# 2. Failure case: return stderr (and stdout if present)
error_msg = f"Error (Exit Code {result.returncode})"
if filtered_stderr:
error_msg += f":\n{filtered_stderr}"
elif stderr_raw:
error_msg += f":\n{stderr_raw}" # Fallback to raw stderr if filtered is empty
if output:
return f"{output}\n\n[{error_msg}]"
return error_msg
def ensure_node_in_path():
"""Ensure node executable is in PATH."""
if not shutil.which("node"):
# Try to infer from GEMINI_PATH
if "/.nvm/" in GEMINI_PATH:
bin_dir = os.path.dirname(GEMINI_PATH)
os.environ["PATH"] = f"{bin_dir}:{os.environ.get('PATH', '')}"
print(f"Added {bin_dir} to PATH")
@mcp.tool()
def run_gemini(
prompt: str,
model: str = "",
sandbox: bool = False,
yolo: bool = True,
output_format: str = "text",
debug: bool = False,
include_directories: str = ""
) -> str:
"""
Execute a command or query using the Gemini CLI.
"""
ensure_node_in_path() # Ensure environment is correct
cmd = [GEMINI_PATH, prompt, "-o", output_format]
if model:
cmd.extend(["-m", model])
if sandbox:
cmd.append("-s")
if yolo:
cmd.append("-y")
if debug:
cmd.append("-d")
if include_directories:
cmd.extend(["--include-directories", include_directories])
try:
# Explicitly pass env to ensure PATH is propagated if modified
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8',
env=os.environ
)
return process_gemini_output(result)
except FileNotFoundError:
return f"Error: Executable not found at {GEMINI_PATH}. PATH={os.environ.get('PATH')}"
except Exception as e:
return f"Error executing Gemini CLI: {str(e)}"
# ... (Include other tools similarly updated) ...
if __name__ == "__main__":
print(f"Starting Gemini CLI MCP Server on {HOST}:{PORT}")
ensure_node_in_path() # Check on startup
mcp.run(transport="sse")
