gemini-bridge

Instructions

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.

Implementation Reference

• src/mcp_server.py:393-419 (handler)

  The @mcp.tool()-decorated handler function implementing 'consult_gemini_with_files'. Includes input schema via type hints and docstring, performs basic validation, and delegates to core execution helper.

  @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:293-370 (helper)

  Core helper function that orchestrates Gemini CLI execution with file attachments. Handles mode selection (inline/at_command), prepares input payload, runs subprocess, and formats output with warnings.

  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:163-215 (helper)

  Helper for preparing inline file content payloads, including safeguards for size limits, truncation, and error handling for file reading.

  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:217-239 (helper)

  Helper for preparing @path prompt directives for Gemini CLI's native file resolution mode.

  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

• src/mcp_server.py:393-393 (registration)

  MCP tool registration decorator for consult_gemini_with_files.

  @mcp.tool()