gemini-bridge

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• src/mcp_server.py:377-403 (handler)

  The primary handler function for the 'consult_gemini_with_files' tool, decorated with @mcp.tool() for registration. It validates the 'files' parameter and delegates execution to the helper function execute_gemini_with_files.
  @mcp.tool() def consult_gemini_with_files( query: str, directory: str, files: list[str] | None = None, model: str | None = None, timeout_seconds: int | None = None, mode: str = "inline", ) -> str: """Send a query to the Gemini CLI with file context. Args: query: Prompt text forwarded to the CLI. directory: Working directory used for resolving relative file paths. files: Relative or absolute file paths to include alongside the prompt. model: Optional model alias (``flash``, ``pro``) or full Gemini model id. timeout_seconds: Optional per-call timeout override in seconds. mode: ``"inline"`` streams truncated snippets; ``"at_command"`` emits ``@path`` directives so Gemini CLI resolves files itself. Returns: Gemini's response or an explanatory error string with any warnings. """ if not files: return "Error: files parameter is required for consult_gemini_with_files" return execute_gemini_with_files(query, directory, files, model, timeout_seconds, mode)
• src/mcp_server.py:377-377 (registration)

  The @mcp.tool() decorator registers the consult_gemini_with_files function as an MCP tool.
  @mcp.tool()
• src/mcp_server.py:277-354 (helper)

  Core helper function that implements the logic for executing the Gemini CLI with file attachments, supporting 'inline' (content snippets) and 'at_command' (@path directives) modes. Prepares input payload and runs subprocess.
  def execute_gemini_with_files( query: str, directory: str = ".", files: Optional[List[str]] = None, model: Optional[str] = None, timeout_seconds: Optional[int] = None, mode: str = "inline", ) -> str: """ Execute gemini CLI command with file attachments. Args: query: The prompt to send to Gemini directory: Working directory for the command files: List of file paths to attach (relative to directory) model: Optional model name (flash, pro, etc.) Returns: CLI output or error message """ # Check if gemini CLI is available if not shutil.which("gemini"): return "Error: Gemini CLI not found. Install with: npm install -g @google/gemini-cli" # Validate directory if not os.path.isdir(directory): return f"Error: Directory does not exist: {directory}" # Validate files parameter if not files: return "Error: No files provided for file attachment mode" # Build command - use stdin for input to avoid hanging selected_model = _normalize_model_name(model) cmd = ["gemini", "-m", selected_model] mode_normalized = mode.lower() warnings: List[str] if mode_normalized not in {"inline", "at_command"}: return f"Error: Unsupported files mode '{mode}'. Use 'inline' or 'at_command'." if mode_normalized == "inline": inline_payload, warnings = _prepare_inline_payload(directory, files) stdin_pieces = [piece for piece in [inline_payload, query] if piece] stdin_content = "\n\n".join(stdin_pieces) else: at_prompt, warnings = _prepare_at_command_prompt(directory, files) stdin_pieces = [piece for piece in [at_prompt, query] if piece] stdin_content = "\n\n".join(stdin_pieces) # Execute CLI command - simple timeout, no retries timeout = _coerce_timeout(timeout_seconds) try: result = subprocess.run( cmd, cwd=directory, capture_output=True, text=True, timeout=timeout, input=stdin_content ) if result.returncode == 0: output = result.stdout.strip() if result.stdout.strip() else "No output from Gemini CLI" else: output = f"Gemini CLI Error: {result.stderr.strip()}" if warnings: warning_block = "Warnings:\n" + "\n".join(f"- {w}" for w in warnings) return f"{warning_block}\n\n{output}" return output except subprocess.TimeoutExpired: return f"Error: Gemini CLI command timed out after {timeout} seconds" except Exception as e: return f"Error executing Gemini CLI: {str(e)}"
• src/mcp_server.py:147-199 (helper)

  Helper for preparing inline file content payload: reads and truncates files, formats into delimited blocks, tracks byte limits and warnings.
  def _prepare_inline_payload(directory: str, files: List[str]) -> Tuple[str, List[str]]: """Return stdin payload for inline mode and any warnings.""" warnings: List[str] = [] file_blocks: List[str] = [] total_bytes = 0 processed = 0 if MAX_INLINE_FILE_COUNT <= 0: warnings.append("Inline attachments disabled via MAX_INLINE_FILE_COUNT<=0") return "", warnings for original_path in files: abs_path, rel_path = _resolve_path(directory, original_path) display_name = rel_path or os.path.basename(abs_path) if not os.path.exists(abs_path): warnings.append(f"Skipped missing file: {display_name}") continue if processed >= MAX_INLINE_FILE_COUNT: warnings.append( f"Inline file limit reached ({MAX_INLINE_FILE_COUNT}); skipped remaining attachments", ) break try: content, truncated, bytes_used = _read_file_for_inline(abs_path) except Exception as exc: # IOError or decoding issues warnings.append(f"Error reading {display_name}: {exc}") continue if total_bytes + bytes_used > MAX_INLINE_TOTAL_BYTES: warnings.append( f"Inline payload exceeded {MAX_INLINE_TOTAL_BYTES} bytes; skipped {display_name} and remaining attachments", ) break block_header = f"=== {display_name} ===" if truncated: block_header += "\n[gemini-bridge] Content truncated for inline transfer" file_blocks.append(f"{block_header}\n{content}") if truncated: warnings.append( f"Truncated {display_name}; only the first {INLINE_CHUNK_HEAD_BYTES}B and last {INLINE_CHUNK_TAIL_BYTES}B were sent", ) total_bytes += bytes_used processed += 1 payload = "\n\n".join(file_blocks) return payload, warnings
• src/mcp_server.py:201-223 (helper)

  Helper for 'at_command' mode: generates @relpath prompt lines for Gemini CLI to resolve files itself.
  def _prepare_at_command_prompt(directory: str, files: List[str]) -> Tuple[str, List[str]]: """Return prompt lines for @-command usage and warnings.""" warnings: List[str] = [] prompt_lines: List[str] = [] for original_path in files: abs_path, rel_path = _resolve_path(directory, original_path) if not os.path.exists(abs_path): warnings.append(f"Skipped missing file: {original_path}") continue if rel_path is None: warnings.append( f"Skipped file outside working directory: {original_path}", ) continue prompt_lines.append(f"@{rel_path}") if not prompt_lines: warnings.append("No readable files resolved for @ command; prompt unchanged") prompt = "\n".join(prompt_lines) return prompt, warnings