gemini.py•2.74 kB
"""Gemini CLI adapter."""
import asyncio
import subprocess
from typing import Any
from .base import CLIAdapter
class GeminiAdapter(CLIAdapter):
"""Adapter for Gemini CLI."""
async def execute(self, task: str, progress_callback: Any = None, timeout: int | None = None, **kwargs: Any) -> tuple[str, str, int]:
"""Execute task using Gemini CLI with optional streaming."""
# Use streaming method from base class if progress_callback provided
if progress_callback:
return await self.execute_streaming(task, progress_callback, timeout, **kwargs)
# Otherwise use buffered execution (legacy)
cmd = self.format_task(task, **kwargs)
resolved_cmd = self.resolve_command(cmd)
process = await asyncio.create_subprocess_exec(
*resolved_cmd,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
env={**subprocess.os.environ, **self.get_env()},
)
try:
effective_timeout = timeout or self.get_timeout()
stdout, stderr = await asyncio.wait_for(
process.communicate(), timeout=effective_timeout
)
return (
stdout.decode("utf-8", errors="replace"),
stderr.decode("utf-8", errors="replace"),
process.returncode or 0,
)
except asyncio.TimeoutError:
process.kill()
await process.wait()
raise TimeoutError(f"Gemini CLI timed out after {effective_timeout}s")
def validate(self) -> bool:
"""Validate Gemini CLI is available."""
try:
subprocess.run(
["which", "gemini"] if subprocess.os.name != "nt" else ["where", "gemini"],
capture_output=True,
check=True,
)
return True
except subprocess.CalledProcessError:
return False
def format_task(self, task: str, **kwargs: Any) -> list[str]:
"""Format task for Gemini CLI."""
cmd = ["gemini"]
# Add model if specified in kwargs or config
if model := kwargs.get("model") or self.config.get("model"):
cmd.extend(["-m", model])
# Add allowed tools if specified
if tools := kwargs.get("allowed_tools"):
tool_list = tools if isinstance(tools, str) else ",".join(tools)
cmd.extend(["--allowed-tools", tool_list])
# Add any custom args from config
for arg in self.get_args():
if arg not in cmd: # Avoid duplicates
cmd.append(arg)
# Add task as final argument (quoted)
cmd.append(task)
return cmd