Create github_client.py

github_client.py

@@ -0,0 +1,187 @@
+"""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"]