github_mcp.py

"""
GitHub MCP Integration — Autonomous Repository Operations
==========================================================
Create repos, commit files, open PRs via GitHub REST API.
"""

import json, os, base64
from typing import Dict
from pathlib import Path
import httpx


class GitHubMCP:
    def __init__(self):
        self.token = os.environ.get("GITHUB_TOKEN") or os.environ.get("HF_TOKEN")
        self.base_url = "https://api.github.com"
        self.headers = {"Accept": "application/vnd.github+json", "X-GitHub-Api-Version": "2022-11-28"}
        if self.token:
            self.headers["Authorization"] = f"Bearer {self.token}"

    async def create_and_push(self, repo_name: str, files: Dict[str, str], description: str = "", private: bool = False) -> Dict:
        result = {"success": False, "repo_url": None, "files_committed": [], "error": None}
        if not self.token:
            local_dir = Path("/app/wizard-vibe/generated") / repo_name
            local_dir.mkdir(parents=True, exist_ok=True)
            for path, content in files.items():
                fp = local_dir / path; fp.parent.mkdir(parents=True, exist_ok=True); fp.write_text(content)
                result["files_committed"].append(str(fp))
            result.update({"success": True, "local_path": str(local_dir), "note": "No GITHUB_TOKEN — saved locally"})
            return result

        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                user_resp = await client.get(f"{self.base_url}/user", headers=self.headers)
                owner = user_resp.json()["login"]

                create = await client.post(f"{self.base_url}/user/repos", headers=self.headers, json={"name": repo_name, "description": description or "Wizard-Vibe Studio", "private": private, "auto_init": True})
                if create.status_code == 201:
                    result["repo_url"] = create.json()["html_url"]
                elif create.status_code == 422:
                    repos = (await client.get(f"{self.base_url}/user/repos", headers=self.headers)).json()
                    for r in repos:
                        if r["name"] == repo_name:
                            result["repo_url"] = r["html_url"]; break
                result["success"] = True

                for path, content in files.items():
                    resp = await client.put(
                        f"{self.base_url}/repos/{owner}/{repo_name}/contents/{path}",
                        headers=self.headers,
                        json={"message": f"✨ Wizard-Vibe: Add {path}", "content": base64.b64encode(content.encode()).decode()},
                    )
                    if resp.status_code in (201, 200):
                        result["files_committed"].append(path)
        except Exception as e:
            result["error"] = str(e)
        return result

    async def push_file(self, repo_name: str, path: str, content: str) -> Dict:
        if not self.token: return {"success": False, "error": "No token"}
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                user_resp = await client.get(f"{self.base_url}/user", headers=self.headers)
                owner = user_resp.json()["login"]
                resp = await client.put(
                    f"{self.base_url}/repos/{owner}/{repo_name}/contents/{path}",
                    headers=self.headers,
                    json={"message": f"✨ Wizard-Vibe: Add {path}", "content": base64.b64encode(content.encode()).decode()},
                )
                return {"success": resp.status_code in (201, 200)}
        except Exception as e:
            return {"success": False, "error": str(e)}

    async def create_pr(self, repo_name: str, title: str, body: str = "", head: str = "wizard-vibe", base: str = "main") -> Dict:
        if not self.token: return {"success": False, "error": "No token"}
        try:
            async with httpx.AsyncClient(timeout=30.0) as client:
                user_resp = await client.get(f"{self.base_url}/user", headers=self.headers)
                owner = user_resp.json()["login"]
                resp = await client.post(
                    f"{self.base_url}/repos/{owner}/{repo_name}/pulls",
                    headers=self.headers,
                    json={"title": title, "body": body or "PR by Wizard-Vibe", "head": head, "base": base},
                )
                if resp.status_code == 201:
                    d = resp.json()
                    return {"success": True, "pr_url": d["html_url"], "pr_number": d["number"]}
                return {"success": False, "error": resp.text}
        except Exception as e:
            return {"success": False, "error": str(e)}