consult_gemini_with_files
Send queries to Gemini AI with file context for analysis, using the official CLI to process documents alongside prompts.
Instructions
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.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | ||
| directory | Yes | ||
| files | No | ||
| model | No | ||
| timeout_seconds | No | ||
| mode | No | inline |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| result | Yes |
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()