backend/tools/git_tools.py

# backend/tasks/tools/git_tools.py
import asyncio
from .terminal_tools import run_command
import json
import urllib.request

async def git_status() -> str:
    res = await run_command("git status")
    return res.get("stdout", "") + res.get("stderr", "")

async def git_commit(message: str) -> str:
    res = await run_command(f'git commit -m "{message}"')
    return res.get("stdout", "") + res.get("stderr", "")

async def git_push() -> str:
    res = await run_command("git push")
    return res.get("stdout", "") + res.get("stderr", "")

async def create_pr(title: str, body: str) -> dict:
    """Create a PR via GitHub API."""
    # Requires GITHUB_TOKEN to be available in env or vault
    import os
    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        return {"error": "GITHUB_TOKEN not set"}
        
    # We would need to determine the repo owner/name, either passed in or parsed from `git remote -v`
    remote_res = await run_command("git config --get remote.origin.url")
    remote_url = remote_res.get("stdout", "").strip()
    
    # Simple parse for github.com:owner/repo.git or https://github.com/owner/repo.git
    owner_repo = ""
    if "github.com" in remote_url:
        parts = remote_url.replace("git@github.com:", "").replace("https://github.com/", "").replace(".git", "").split("/")
        if len(parts) >= 2:
            owner_repo = f"{parts[-2]}/{parts[-1]}"
            
    if not owner_repo:
        return {"error": "Could not determine GitHub repository"}
        
    branch_res = await run_command("git rev-parse --abbrev-ref HEAD")
    head_branch = branch_res.get("stdout", "").strip()
    
    # We assume base is main or master
    base_branch = "main"
    
    url = f"https://api.github.com/repos/{owner_repo}/pulls"
    data = json.dumps({
        "title": title,
        "body": body,
        "head": head_branch,
        "base": base_branch
    }).encode("utf-8")
    
    req = urllib.request.Request(url, data=data, method="POST")
    req.add_header("Authorization", f"Bearer {token}")
    req.add_header("Accept", "application/vnd.github.v3+json")
    
    try:
        def _post():
            with urllib.request.urlopen(req) as response:
                return json.loads(response.read().decode('utf-8'))
        return await asyncio.to_thread(_post)
    except Exception as e:
        return {"error": str(e)}