github_client.py

"""Lightweight async GitHub API client used by the MCP server."""
from __future__ import annotations

import base64
import binascii
import os
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional

import httpx


class GitHubAPIError(RuntimeError):
    """Raised when the GitHub API returns an error response."""


@dataclass(slots=True)
class GitHubClient:
    """Thin wrapper around the GitHub REST API."""

    token: Optional[str] = None
    base_url: str = "https://api.github.com"
    timeout: float = 15.0

    def __post_init__(self) -> None:
        if not self.token:
            self.token = os.environ.get("GITHUB_TOKEN") or os.environ.get("GITHUB_ACCESS_TOKEN")
        self.base_url = self.base_url.rstrip("/")

    def _headers(self) -> Dict[str, str]:
        headers = {
            "Accept": "application/vnd.github+json",
            "User-Agent": "github-chatgpt-app",
        }
        if self.token:
            headers["Authorization"] = f"Bearer {self.token}"
        return headers

    async def _get(self, endpoint: str, params: Optional[Dict[str, Any]] = None) -> Any:
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.get(url, headers=self._headers(), params=params)
            response.raise_for_status()
            return response.json()
        except httpx.HTTPStatusError as exc:
            detail = exc.response.json() if exc.response.content else {"message": exc.response.text}
            message = detail.get("message", "GitHub API error")
            raise GitHubAPIError(f"{message} (status {exc.response.status_code})") from exc
        except httpx.RequestError as exc:
            raise GitHubAPIError(f"Unable to reach GitHub: {exc}") from exc

    async def get_repo_info(self, owner: str, repo: str) -> Dict[str, Any]:
        return await self._get(f"repos/{owner}/{repo}")

    async def list_issues(
        self,
        owner: str,
        repo: str,
        state: str = "open",
        labels: Optional[Iterable[str]] = None,
        limit: int = 25,
        include_pull_requests: bool = False,
    ) -> List[Dict[str, Any]]:
        params: Dict[str, Any] = {
            "state": state,
            "per_page": min(max(limit, 1), 100),
        }
        if labels:
            params["labels"] = ",".join(labels)
        data = await self._get(f"repos/{owner}/{repo}/issues", params=params)
        if include_pull_requests:
            return data
        return [issue for issue in data if "pull_request" not in issue]

    async def list_pull_requests(
        self,
        owner: str,
        repo: str,
        state: str = "open",
        limit: int = 25,
    ) -> List[Dict[str, Any]]:
        params = {
            "state": state,
            "per_page": min(max(limit, 1), 100),
        }
        return await self._get(f"repos/{owner}/{repo}/pulls", params=params)

    async def get_readme(self, owner: str, repo: str, ref: Optional[str] = None) -> Dict[str, Any]:
        params = {"ref": ref} if ref else None
        try:
            data = await self._get(f"repos/{owner}/{repo}/readme", params=params)
        except GitHubAPIError as exc:
            message = str(exc)
            if "status 404" in message:
                return {
                    "name": None,
                    "path": None,
                    "sha": None,
                    "content": "",
                    "error": "README not found",
                }
            raise
        raw_content = data.get("content", "") or ""
        encoding = (data.get("encoding") or "base64").lower()
        decoded = ""
        error: Optional[str] = None
        if encoding == "base64":
            try:
                decoded_bytes = base64.b64decode(raw_content, validate=False)
                decoded = decoded_bytes.decode("utf-8", errors="replace")
            except (binascii.Error, UnicodeDecodeError):
                error = "README content could not be decoded from base64."
        else:
            if isinstance(raw_content, str):
                decoded = raw_content
            else:
                try:
                    decoded = raw_content.decode("utf-8", errors="replace")
                except Exception:
                    decoded = str(raw_content)
                    error = f"Unsupported README encoding: {encoding}"
        return {
            "name": data.get("name"),
            "path": data.get("path"),
            "sha": data.get("sha"),
            "content": decoded,
            "error": error,
        }

    async def list_contributors(
        self,
        owner: str,
        repo: str,
        limit: int = 25,
        include_anonymous: bool = False,
    ) -> List[Dict[str, Any]]:
        params = {
            "per_page": min(max(limit, 1), 100),
            "anon": str(include_anonymous).lower(),
        }
        return await self._get(f"repos/{owner}/{repo}/contributors", params=params)

    async def list_files(
        self,
        owner: str,
        repo: str,
        ref: str = "main",
        directory: Optional[str] = None,
        limit: int = 200,
    ) -> List[Dict[str, Any]]:
        params = {"recursive": "1"}
        data = await self._get(f"repos/{owner}/{repo}/git/trees/{ref}", params=params)
        tree = data.get("tree", [])
        if directory:
            normalized = directory.strip("/")
            tree = [node for node in tree if node.get("path", "").startswith(normalized)]
        limited = tree[: min(max(limit, 1), len(tree))]
        return [
            {
                "path": node.get("path"),
                "type": node.get("type"),
                "size": node.get("size"),
                "sha": node.get("sha"),
            }
            for node in limited
        ]

    async def search_repositories(
        self,
        query: str,
        sort: Optional[str] = None,
        order: str = "desc",
        limit: int = 25,
    ) -> List[Dict[str, Any]]:
        params: Dict[str, Any] = {
            "q": query,
            "per_page": min(max(limit, 1), 100),
            "order": order,
        }
        if sort:
            params["sort"] = sort
        data = await self._get("search/repositories", params=params)
        return data.get("items", [])


__all__ = ["GitHubClient", "GitHubAPIError"]