gemini_mcp

Instructions

Invokes Gemini via ACP (Agent Client Protocol) for AI-driven tasks.

**Return structure:**
    - `success`: boolean indicating execution status
    - `SESSION_ID`: ACP session identifier (auto-managed per workspace)
    - `agent_messages`: concatenated assistant response text
    - `thought`: agent reasoning/thinking (when available)
    - `stop_reason`: why the agent stopped (end_turn, max_tokens, etc.)
    - `tool_calls`: list of tool invocations made by the agent (if any)
    - `plan`: agent execution plan entries (if any)
    - `error`: error description when `success=False`

**Best practices:**
    - Sessions auto-reuse per workspace with turn-count eviction
    - ALWAYS pass `model`. Use `gemini-3.1-pro-preview` for complex tasks, `gemini-3-flash-preview` for simple tasks
    - Use `approval_mode` to control tool approval: yolo (default), auto_edit, default, plan
    - On 429 capacity errors, automatically retries with `gemini-3-flash-preview`
    - Pass `image_path` for vision analysis (requires agent image support)
    - Pass `context` to inject text as embedded resource (ACP resource ContentBlock)
    - Pass `allowed_mcp_servers` to filter which MCP servers Gemini loads

Implementation Reference

• src/geminimcp/server.py:795-1036 (handler)

  Handler for the 'gemini' MCP tool, which acts as a bridge to the Gemini CLI. It manages sessions, invokes the CLI, handles streaming responses, and implements retries/fallbacks.

  @mcp.tool(
      name="gemini",
      annotations=ToolAnnotations(
          title="Gemini CLI Agent",
          readOnlyHint=False,
          destructiveHint=True,
          idempotentHint=False,
          openWorldHint=True,
      ),
      description="""
      Invokes Gemini via ACP (Agent Client Protocol) for AI-driven tasks.
  
      **Return structure:**
          - `success`: boolean indicating execution status
          - `SESSION_ID`: ACP session identifier (auto-managed per workspace)
          - `agent_messages`: concatenated assistant response text
          - `thought`: agent reasoning/thinking (when available)
          - `stop_reason`: why the agent stopped (end_turn, max_tokens, etc.)
          - `tool_calls`: list of tool invocations made by the agent (if any)
          - `plan`: agent execution plan entries (if any)
          - `error`: error description when `success=False`
  
      **Best practices:**
          - Sessions auto-reuse per workspace with turn-count eviction
          - ALWAYS pass `model`. Use `gemini-3.1-pro-preview` for complex tasks, `gemini-3-flash-preview` for simple tasks
          - Use `approval_mode` to control tool approval: yolo (default), auto_edit, default, plan
          - On 429 capacity errors, automatically retries with `gemini-3-flash-preview`
          - Pass `image_path` for vision analysis (requires agent image support)
          - Pass `context` to inject text as embedded resource (ACP resource ContentBlock)
          - Pass `allowed_mcp_servers` to filter which MCP servers Gemini loads
      """,
  )
  async def gemini(
      PROMPT: Annotated[
          str,
          Field(description="Instruction for the task to send to Gemini."),
      ],
      cd: Annotated[
          Path,
          Field(
              description="Set the workspace root for Gemini before executing the task."
          ),
      ],
      model: Annotated[
          str,
          Field(
              description="REQUIRED. Pass 'gemini-3.1-pro-preview' for complex tasks, "
              "'gemini-3-flash-preview' for simple tasks."
          ),
      ] = "gemini-3.1-pro-preview",
      approval_mode: Annotated[
          str,
          Field(
              description="Tool approval mode. "
              "'yolo': auto-approve all (default). "
              "'auto_edit': auto-approve edits only. "
              "'default': prompt for every action (safest). "
              "'plan': read-only mode."
          ),
      ] = "yolo",
      image_path: Annotated[
          str,
          Field(
              description="Path to an image file for vision analysis. "
              "Sent as image ContentBlock. Empty string means no image."
          ),
      ] = "",
      context: Annotated[
          str,
          Field(
              description="Text context to inject as ACP resource ContentBlock. "
              "Use for passing file contents, docs, or background info that Gemini should reference."
          ),
      ] = "",
      allowed_mcp_servers: Annotated[
          Optional[List[str]],
          Field(
              description="Filter which MCP servers Gemini loads. "
              "Pass a list of server names to include. None means load all discovered servers."
          ),
      ] = None,
  ) -> Dict[str, Any]:
      """Execute a Gemini session via ACP and return results."""
      if not shutil.which("gemini"):
          return {"success": False, "error": "CLI tool 'gemini' not found in PATH."}
  
      if not cd.exists():
          return {
              "success": False,
              "error": f"Workspace directory `{cd.absolute().as_posix()}` does not exist.",
          }
  
      if approval_mode not in _APPROVAL_MODES:
          return {
              "success": False,
              "error": f"Invalid approval_mode '{approval_mode}'. "
              f"Valid values: {', '.join(_APPROVAL_MODES.keys())}",
          }
  
      cwd = cd.absolute().as_posix()
      result = _bridge.prompt(
          cwd,
          PROMPT,
          model=model,
          approval_mode=approval_mode,
          image_path=image_path,
          context=context,
          allowed_mcp_servers=allowed_mcp_servers,
      )
  
      # Session error → retry with fresh session
      if not result["success"] and result.get("SESSION_ID"):
          _bridge._sessions.pop(cwd, None)
          result = _bridge.prompt(
              cwd,
              PROMPT,
              model=model,
              approval_mode=approval_mode,
              image_path=image_path,
              context=context,
              allowed_mcp_servers=allowed_mcp_servers,
          )
  
      # 429 fallback: capacity error → retry with flash model
      # Skip fallback for auto-* models (they handle routing internally)
      _FALLBACK_MODEL = "gemini-3-flash-preview"
      if (
          not result["success"]
          and model != _FALLBACK_MODEL
          and not model.startswith("auto-")
          and any(
              kw in result.get("error", "").lower()
              for kw in ("capacity", "429", "resource_exhausted", "overloaded")
          )
      ):
          _bridge._sessions.pop(cwd, None)
          result = _bridge.prompt(
              cwd, PROMPT, model=_FALLBACK_MODEL, approval_mode=approval_mode
          )
          result["fallback_model"] = _FALLBACK_MODEL
  
      return result
  
  
  @mcp.tool(
      name="list_models",
      annotations=ToolAnnotations(
          title="List Available Models",
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=False,
      ),
      description="List available Gemini models and current bridge state. "
      "Returns known models, current active model, and agent info.",
  )
  async def list_models() -> Dict[str, Any]:
      """List available models and bridge status."""
      return {
          "models": _KNOWN_MODELS,
          "approval_modes": list(_APPROVAL_MODES.keys()),
          "current_model": _bridge._current_model or "(not started)",
          "agent_info": _bridge._agent_info or None,
          "bridge_version": VERSION,
          "process_running": _bridge._proc is not None and _bridge._proc.poll() is None,
      }
  
  
  @mcp.tool(
      name="list_sessions",
      annotations=ToolAnnotations(
          title="List Active Sessions",
          readOnlyHint=True,
          destructiveHint=False,
          idempotentHint=True,
          openWorldHint=False,
      ),
      description="List all active ACP sessions managed by the bridge. "
      "Shows workspace path, session ID, turn count, and model for each session.",
  )
  async def list_sessions() -> Dict[str, Any]:
      """List active sessions."""
      sessions = []
      for workspace, info in _bridge._sessions.items():
          sessions.append(
              {
                  "workspace": workspace,
                  "session_id": info["session_id"],
                  "turn_count": info["turn_count"],
                  "max_turns": _MAX_TURNS_PER_SESSION,
                  "model": info.get("actual_model", ""),
              }
          )
      return {
          "sessions": sessions,
          "count": len(sessions),
      }
  
  
  @mcp.tool(
      name="reset_session",
      annotations=ToolAnnotations(
          title="Reset Session",
          readOnlyHint=False,
          destructiveHint=True,
          idempotentHint=True,
          openWorldHint=False,
      ),
      description="Reset (clear) the ACP session for a workspace. "
      "The next gemini call for this workspace will create a fresh session. "
      "Pass workspace path, or omit to reset all sessions.",
  )
  async def reset_session(
      workspace: Annotated[
          str,
          Field(description="Workspace path to reset. Empty string resets all sessions."),
      ] = "",
  ) -> Dict[str, Any]:
      """Reset session for a workspace or all sessions."""
      if workspace:
          removed = _bridge._sessions.pop(workspace, None)
          if not removed:
              # Try matching by suffix (user might pass partial path)
              matched = [k for k in _bridge._sessions if k.endswith(workspace)]
              if matched:
                  for k in matched:
                      _bridge._sessions.pop(k)
                  return {"reset": matched, "count": len(matched)}
              return {
                  "reset": [],
                  "count": 0,
                  "message": "No session found for workspace",
              }
          return {"reset": [workspace], "count": 1}
      else:
          count = len(_bridge._sessions)
          _bridge._sessions.clear()
          return {"reset": "all", "count": count}
  
  
  def run() -> None:
      """Start the MCP server over stdio transport."""

• src/geminimcp/server.py:568-568 (handler)

  The internal `AcpBridge.prompt` method which handles the communication with the underlying Gemini --acp subprocess.

  def prompt(