"""Send prompts to Gemini CLI for sub-agent code tasks (UNSANDBOXED_EXEC)."""
from __future__ import annotations
import asyncio
import json
import logging
import os
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING
from tools.alter_privileges import PRIVILEGES, has_privilege
if TYPE_CHECKING:
from tool_context import ToolContext
logger = logging.getLogger(__name__)
TOOL_NAME = "send_gemini_prompt"
TOOL_DESCRIPTION = (
"Send a prompt to Gemini CLI to perform development/ops tasks. "
"Uses AI-powered code editing. Requires UNSANDBOXED_EXEC (same tier as Cursor CLI)."
)
TOOL_PARAMETERS = {
"type": "object",
"properties": {
"prompt": {
"type": "string",
"description": "The prompt describing what changes to make or what to query.",
},
"timeout": {
"type": "integer",
"description": "Max seconds to wait for Gemini to respond (default 300, max 600).",
},
"working_directory": {
"type": "string",
"description": "Directory to run the command in (default '/sandbox').",
},
},
"required": ["prompt"],
}
[docs]
async def run(
prompt: str,
timeout: int = 300,
working_directory: str = "/sandbox",
ctx: ToolContext | None = None,
) -> str:
"""Execute this tool and return the result.
Args:
prompt (str): The prompt value.
timeout (int): Maximum wait time in seconds.
working_directory (str): The working directory value.
ctx (ToolContext | None): Tool execution context providing access to bot internals.
Returns:
str: Result string.
"""
user_id = getattr(ctx, "user_id", "") or ""
redis = getattr(ctx, "redis", None)
config = getattr(ctx, "config", None)
if not await has_privilege(redis, user_id, PRIVILEGES["UNSANDBOXED_EXEC"], config):
return json.dumps({
"success": False,
"error": (
"The user does not have the UNSANDBOXED_EXEC privilege. "
"Ask an admin to grant it with the alter_privileges tool."
),
})
if not prompt or not prompt.strip():
return json.dumps({"success": False, "error": "Empty prompt provided."})
if timeout > 600:
timeout = 600
logger.warning("Timeout limited to 600 seconds for user %s", user_id)
def _gemini_cwd_ok(raw: str) -> tuple[str | None, str | None]:
try:
p = Path(raw).resolve()
except (OSError, RuntimeError):
return None, "Invalid working_directory path."
allow_roots: list[Path] = []
for root in ("/sandbox", os.getcwd()):
try:
allow_roots.append(Path(root).resolve())
except (OSError, RuntimeError):
continue
for root in allow_roots:
try:
if p == root or p.is_relative_to(root):
return str(p), None
except ValueError:
continue
return None, (
"working_directory must be under /sandbox or the process "
"current working directory."
)
wd_resolved, wd_err = _gemini_cwd_ok(working_directory or "/sandbox")
if wd_err:
return json.dumps({"success": False, "error": wd_err})
working_directory = wd_resolved
if not await asyncio.to_thread(os.path.exists, working_directory):
try:
await asyncio.to_thread(os.makedirs, working_directory, exist_ok=True)
logger.info("Created working directory: %s", working_directory)
except Exception as e:
return json.dumps({
"success": False,
"error": f"Failed to create working directory '{working_directory}': {e}",
})
if not await asyncio.to_thread(os.path.isdir, working_directory):
return json.dumps({
"success": False,
"error": f"Working directory '{working_directory}' is not a directory.",
})
logger.info(
"Executing gemini CLI for user %s in %s: %s...",
user_id, working_directory, prompt[:100],
)
try:
cmd_args = ["gemini", "-y", prompt]
process = await asyncio.create_subprocess_exec(
*cmd_args,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
cwd=working_directory,
)
stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=timeout)
stdout_text = stdout.decode(errors="replace") if stdout else ""
stderr_text = stderr.decode(errors="replace") if stderr else ""
response_data = None
if stdout_text.strip():
try:
response_data = json.loads(stdout_text)
except json.JSONDecodeError:
response_data = {"raw_output": stdout_text}
result = {
"success": process.returncode == 0,
"return_code": process.returncode,
"working_directory": working_directory,
"response": response_data,
}
if stderr_text.strip():
result["stderr"] = stderr_text.strip()
logger.info("Gemini CLI completed for user %s, return code: %s", user_id, process.returncode)
return json.dumps(result, indent=2, ensure_ascii=False)
except asyncio.TimeoutError:
logger.warning("Gemini CLI timed out for user %s: %s...", user_id, prompt[:50])
return json.dumps({
"success": False,
"error": f"Gemini CLI timed out after {timeout} seconds. Complex changes may require more time.",
})
except FileNotFoundError:
logger.error("gemini command not found")
return json.dumps({
"success": False,
"error": "gemini command not found. Is Gemini CLI installed and in PATH?",
})
except Exception as e:
logger.error("Gemini CLI error for user %s: %s", user_id, e, exc_info=True)
return json.dumps({"success": False, "error": f"An error occurred: {e}"})