"""
SSE Transport wrapper for Gemini CLI MCP Server.
Exposes the MCP server over HTTP with SSE transport for network access.
"""
import os
from mcp.server.fastmcp import FastMCP
import subprocess
import re
# 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-?]*[ -/]*[@-~])')
return ansi_escape.sub('', text)
def filter_stderr(stderr: str, keep_extensions: bool = False) -> str:
"""
Filter out Gemini CLI noise from stderr, keeping only real errors.
Args:
stderr: The stderr output to filter.
keep_extensions: If True, keep extension list output (for gemini_list_extensions).
Filters out:
- Extension loading messages
- Startup profiler metrics
- YOLO mode notices
- Credential loading messages
- ImportProcessor warnings for non-critical extensions
"""
if not stderr:
return ""
# Patterns to filter out (noise, not real errors)
noise_patterns = [
r'^\[STARTUP\].*$', # Startup profiler messages
r'^Loading extension:.*$', # Extension loading
r'^Loaded cached credentials\.$', # Credential loading
r'^YOLO mode is enabled\..*$', # YOLO mode notice
r'^\[ERROR\] \[ImportProcessor\].*parcel.*$', # Parcel config warnings
]
# Only filter extension list if not keeping them
if not keep_extensions:
noise_patterns.extend([
r'^Installed extensions:$', # Extension list header
r'^- [a-z0-9-]+$', # Extension list items
])
# Compile patterns
compiled_patterns = [re.compile(p, re.MULTILINE) for p in noise_patterns]
# Filter lines
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 Gemini CLI output, handling stdout and stderr appropriately.
Args:
result: The subprocess result to process.
keep_extensions: If True, keep extension list output in stderr.
Returns clean output with only relevant errors included.
"""
# Get stdout
output = strip_ansi(result.stdout).strip() if result.stdout else ""
# Filter and process stderr
filtered_stderr = filter_stderr(strip_ansi(result.stderr), keep_extensions) if result.stderr else ""
# For extension listing, the output is in stderr
if keep_extensions and filtered_stderr and not output:
return filtered_stderr
# Only append stderr if there are real errors
if filtered_stderr:
if output:
output += f"\n\n[ERRORS]\n{filtered_stderr}"
else:
output = f"[ERRORS]\n{filtered_stderr}"
# Check return code for unexpected failures
if result.returncode != 0 and not output:
return f"Error: Gemini CLI exited with code {result.returncode}"
return output if output else "No output returned"
@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.
Args:
prompt: The prompt or command to send to Gemini.
model: Optional model to use (e.g., gemini-2.5-flash, gemini-2.0-flash-exp, gemini-1.5-pro).
sandbox: Whether to run in sandbox mode (isolated environment for safety).
yolo: Whether to auto-approve all actions (recommended for automated use).
output_format: Output format - 'text' (default), 'json', or 'stream-json'.
debug: Enable debug mode for verbose output.
include_directories: Comma-separated list of additional directories to include.
"""
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:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error executing Gemini CLI: {str(e)}"
@mcp.tool()
def gemini_resume(session: str = "latest", prompt: str = "") -> str:
"""
Resume a previous Gemini CLI session.
Args:
session: Session to resume. Use "latest" for most recent or index number.
prompt: Optional prompt to continue the conversation with.
"""
cmd = [GEMINI_PATH, "-r", session, "-o", "text", "-y"]
if prompt:
cmd.append(prompt)
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error resuming session: {str(e)}"
@mcp.tool()
def gemini_list_sessions() -> str:
"""
List available Gemini CLI sessions for the current project.
"""
cmd = [GEMINI_PATH, "--list-sessions"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error listing sessions: {str(e)}"
@mcp.tool()
def gemini_list_extensions() -> str:
"""
List all available Gemini CLI extensions.
"""
cmd = [GEMINI_PATH, "-l"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
# Keep extension output for this command
return process_gemini_output(result, keep_extensions=True)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error listing extensions: {str(e)}"
@mcp.tool()
def gemini_version() -> str:
"""
Get the Gemini CLI version information.
"""
cmd = [GEMINI_PATH, "--version"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error getting version: {str(e)}"
@mcp.tool()
def gemini_delete_session(session_index: str) -> str:
"""
Delete a Gemini CLI session by index number.
Use gemini_list_sessions() first to see available sessions.
Args:
session_index: The index number of the session to delete.
"""
cmd = [GEMINI_PATH, "--delete-session", session_index]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error deleting session: {str(e)}"
@mcp.tool()
def gemini_json(prompt: str, model: str = "", sandbox: bool = False, yolo: bool = True) -> str:
"""
Execute a Gemini prompt and return structured JSON output.
Useful for programmatic parsing of responses.
Args:
prompt: The prompt or command to send to Gemini.
model: Optional model to use.
sandbox: Whether to run in sandbox mode.
yolo: Whether to auto-approve all actions.
"""
cmd = [GEMINI_PATH, prompt, "-o", "json"]
if model:
cmd.extend(["-m", model])
if sandbox:
cmd.append("-s")
if yolo:
cmd.append("-y")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error executing Gemini CLI: {str(e)}"
@mcp.tool()
def gemini_with_context(prompt: str, files: str = "", directories: str = "", model: str = "", yolo: bool = True) -> str:
"""
Execute a Gemini prompt with specific file or directory context.
Uses @ syntax to include file contents in the prompt.
Args:
prompt: The prompt or command to send to Gemini.
files: Comma-separated list of file paths to include (e.g., "src/main.py,README.md").
directories: Comma-separated list of directories to include.
model: Optional model to use.
yolo: Whether to auto-approve all actions.
"""
# Build prompt with @ references
context_parts = []
if files:
for f in files.split(","):
f = f.strip()
if f:
context_parts.append(f"@{f}")
if directories:
for d in directories.split(","):
d = d.strip()
if d:
context_parts.append(f"@{d}")
full_prompt = " ".join(context_parts + [prompt]) if context_parts else prompt
cmd = [GEMINI_PATH, full_prompt, "-o", "text"]
if model:
cmd.extend(["-m", model])
if yolo:
cmd.append("-y")
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error executing Gemini CLI: {str(e)}"
@mcp.tool()
def gemini_shell(command: str) -> str:
"""
Execute a shell command through Gemini CLI using ! syntax.
Gemini can observe the output and provide analysis.
Args:
command: Shell command to execute (without the ! prefix).
"""
# Use ! syntax for shell commands in Gemini
prompt = f"!{command}"
cmd = [GEMINI_PATH, prompt, "-o", "text", "-y"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error executing shell command: {str(e)}"
@mcp.tool()
def gemini_chat_save(name: str = "") -> str:
"""
Save the current Gemini chat session.
Args:
name: Optional name for the saved session.
"""
cmd = [GEMINI_PATH, "/chat save" + (f" {name}" if name else ""), "-o", "text", "-y"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error saving chat: {str(e)}"
@mcp.tool()
def gemini_help() -> str:
"""
Get Gemini CLI help information and available commands.
"""
cmd = [GEMINI_PATH, "--help"]
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
check=False,
encoding='utf-8'
)
return process_gemini_output(result)
except FileNotFoundError:
return "Error: 'gemini' executable not found. Please ensure Gemini CLI is installed and in your PATH."
except Exception as e:
return f"Error getting help: {str(e)}"
if __name__ == "__main__":
print(f"Starting Gemini CLI MCP Server on {HOST}:{PORT}")
print(f"SSE endpoint: http://{HOST}:{PORT}/sse")
print(f"Messages endpoint: http://{HOST}:{PORT}/messages/")
# Try to disable transport security if available (for newer MCP versions)
try:
if hasattr(mcp, 'settings') and mcp.settings is not None:
if hasattr(mcp.settings, 'transport_security') and mcp.settings.transport_security is not None:
mcp.settings.transport_security.enable_dns_rebinding_protection = False
mcp.settings.transport_security.allowed_hosts = ["*"]
mcp.settings.transport_security.allowed_origins = ["*"]
print("Transport security configured for external access")
except Exception as e:
print(f"Note: Could not configure transport security: {e}")
mcp.run(transport="sse")
