copilot.py•2.31 kB
"""GitHub Copilot CLI adapter."""
import asyncio
import subprocess
from typing import Any
from .base import CLIAdapter
class CopilotAdapter(CLIAdapter):
"""Adapter for GitHub Copilot CLI."""
async def execute(self, task: str, progress_callback: Any = None, timeout: int | None = None, **kwargs: Any) -> tuple[str, str, int]:
"""Execute task using Copilot CLI with optional streaming."""
if progress_callback:
return await self.execute_streaming(task, progress_callback, timeout, **kwargs)
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"Copilot CLI timed out after {effective_timeout}s")
def validate(self) -> bool:
"""Validate Copilot CLI is available."""
try:
# Check for copilot CLI
subprocess.run(
["which", "copilot"] if subprocess.os.name != "nt" else ["where", "copilot"],
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 Copilot CLI."""
cmd = ["copilot"]
# Add subcommand (suggest, explain, etc.)
subcommand = kwargs.get("subcommand", "suggest")
cmd.append(subcommand)
# Add any custom args from config
cmd.extend(self.get_args())
# Add task
cmd.append(task)
return cmd