server.py

"""
MCP Server: GitHub Intelligence

Provides Claude with tools to query GitHub's REST API for repository
analytics, issue tracking, pull-request status, contributor insights,
and repository search.

Works without authentication (60 requests/hour).
Set GITHUB_TOKEN in a .env file for higher rate limits (5,000 requests/hour).
"""

from __future__ import annotations

import logging
import os
import sys
from typing import Any, Optional

import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP

# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------

load_dotenv()

GITHUB_API_BASE = "https://api.github.com"
GITHUB_TOKEN: str | None = os.getenv("GITHUB_TOKEN") or None
REQUEST_TIMEOUT = 30.0  # seconds

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    stream=sys.stderr,
)
logger = logging.getLogger("github-intel")

# ---------------------------------------------------------------------------
# HTTP helpers
# ---------------------------------------------------------------------------


def _build_headers() -> dict[str, str]:
    """Return default headers, including Authorization when a token exists."""
    headers: dict[str, str] = {
        "Accept": "application/vnd.github+json",
        "X-GitHub-Api-Version": "2022-11-28",
    }
    if GITHUB_TOKEN:
        headers["Authorization"] = f"Bearer {GITHUB_TOKEN}"
    return headers


async def _github_request(
    method: str,
    path: str,
    *,
    params: dict[str, Any] | None = None,
) -> dict[str, Any] | list[Any]:
    """Execute an authenticated request against the GitHub REST API.

    Handles common error conditions:
    - 401 Unauthorized (bad token)
    - 403 Forbidden / rate-limit exceeded
    - 404 Not Found
    - Unexpected HTTP errors
    - Network / timeout errors

    Returns the parsed JSON body on success.
    """
    url = f"{GITHUB_API_BASE}{path}"

    async with httpx.AsyncClient(timeout=REQUEST_TIMEOUT) as client:
        try:
            response = await client.request(
                method, url, headers=_build_headers(), params=params
            )
        except httpx.TimeoutException:
            raise RuntimeError(
                f"Request to {url} timed out after {REQUEST_TIMEOUT}s. "
                "GitHub may be experiencing issues — try again shortly."
            )
        except httpx.HTTPError as exc:
            raise RuntimeError(f"Network error while contacting GitHub: {exc}")

    # --- Error handling --------------------------------------------------- #
    if response.status_code == 401:
        raise RuntimeError(
            "GitHub returned 401 Unauthorized. "
            "Your GITHUB_TOKEN may be invalid or expired."
        )

    if response.status_code == 403:
        remaining = response.headers.get("x-ratelimit-remaining", "?")
        reset = response.headers.get("x-ratelimit-reset", "?")
        raise RuntimeError(
            f"GitHub returned 403 Forbidden. "
            f"Rate-limit remaining: {remaining}. "
            f"Resets at Unix timestamp: {reset}. "
            "Consider adding a GITHUB_TOKEN for 5,000 requests/hour."
        )

    if response.status_code == 404:
        raise RuntimeError(
            f"GitHub returned 404 Not Found for {path}. "
            "Check that the owner, repo, or resource exists and is public."
        )

    if response.status_code == 422:
        body = response.json()
        msg = body.get("message", "Unprocessable Entity")
        raise RuntimeError(f"GitHub returned 422: {msg}")

    if not response.is_success:
        raise RuntimeError(
            f"GitHub API error {response.status_code}: {response.text[:500]}"
        )

    return response.json()


def _rate_limit_note(response_headers: httpx.Headers | None = None) -> str:
    """Build a human-readable rate-limit footnote (best-effort)."""
    if GITHUB_TOKEN:
        return "Authenticated mode (5,000 requests/hour)."
    return "Unauthenticated mode (60 requests/hour). Set GITHUB_TOKEN for higher limits."


# ---------------------------------------------------------------------------
# FastMCP server
# ---------------------------------------------------------------------------

mcp = FastMCP(
    "GitHub Intelligence",
    description=(
        "Query GitHub repositories, issues, pull requests, contributors, "
        "and search across all of GitHub — directly from Claude."
    ),
)


# ---------------------------------------------------------------------------
# Tool 1 — Repository Info
# ---------------------------------------------------------------------------


@mcp.tool()
async def get_repository_info(owner: str, repo: str) -> dict[str, Any]:
    """Retrieve comprehensive information about a GitHub repository.

    Returns description, star/fork/issue counts, primary language, license,
    creation and last-update timestamps, topics, default branch, size in KB,
    and visibility status.

    Args:
        owner: Repository owner (user or organization).
        repo:  Repository name.
    """
    data: dict[str, Any] = await _github_request("GET", f"/repos/{owner}/{repo}")  # type: ignore[assignment]

    license_info = data.get("license")
    license_name = license_info["spdx_id"] if license_info else None

    return {
        "full_name": data["full_name"],
        "description": data.get("description"),
        "homepage": data.get("homepage"),
        "stars": data["stargazers_count"],
        "forks": data["forks_count"],
        "open_issues": data["open_issues_count"],
        "watchers": data["subscribers_count"],
        "language": data.get("language"),
        "license": license_name,
        "topics": data.get("topics", []),
        "default_branch": data["default_branch"],
        "size_kb": data["size"],
        "visibility": "private" if data["private"] else "public",
        "archived": data["archived"],
        "created_at": data["created_at"],
        "updated_at": data["updated_at"],
        "pushed_at": data["pushed_at"],
        "html_url": data["html_url"],
        "_note": _rate_limit_note(),
    }


# ---------------------------------------------------------------------------
# Tool 2 — Issues
# ---------------------------------------------------------------------------


@mcp.tool()
async def list_issues(
    owner: str,
    repo: str,
    state: str = "open",
    labels: Optional[str] = None,
    limit: int = 10,
) -> dict[str, Any]:
    """List issues for a GitHub repository with optional filtering.

    Args:
        owner:  Repository owner (user or organization).
        repo:   Repository name.
        state:  Filter by state — "open", "closed", or "all". Default: "open".
        labels: Comma-separated label names to filter by (e.g. "bug,help wanted").
        limit:  Maximum number of issues to return (1-100). Default: 10.
    """
    if state not in ("open", "closed", "all"):
        raise ValueError('state must be "open", "closed", or "all".')
    limit = max(1, min(limit, 100))

    params: dict[str, Any] = {
        "state": state,
        "per_page": limit,
        "sort": "created",
        "direction": "desc",
    }
    if labels:
        params["labels"] = labels

    raw: list[dict[str, Any]] = await _github_request(  # type: ignore[assignment]
        "GET", f"/repos/{owner}/{repo}/issues", params=params
    )

    # The /issues endpoint also returns pull requests; filter them out.
    issues = [item for item in raw if "pull_request" not in item]

    results = []
    for issue in issues[:limit]:
        results.append(
            {
                "number": issue["number"],
                "title": issue["title"],
                "state": issue["state"],
                "author": issue["user"]["login"],
                "labels": [label["name"] for label in issue.get("labels", [])],
                "comments": issue["comments"],
                "created_at": issue["created_at"],
                "updated_at": issue["updated_at"],
                "html_url": issue["html_url"],
            }
        )

    return {
        "repository": f"{owner}/{repo}",
        "filter": {"state": state, "labels": labels},
        "total_returned": len(results),
        "issues": results,
        "_note": _rate_limit_note(),
    }


# ---------------------------------------------------------------------------
# Tool 3 — Pull Requests
# ---------------------------------------------------------------------------


@mcp.tool()
async def list_pull_requests(
    owner: str,
    repo: str,
    state: str = "open",
    limit: int = 10,
) -> dict[str, Any]:
    """List pull requests for a GitHub repository.

    Args:
        owner: Repository owner (user or organization).
        repo:  Repository name.
        state: Filter by state — "open", "closed", or "all". Default: "open".
        limit: Maximum number of pull requests to return (1-100). Default: 10.
    """
    if state not in ("open", "closed", "all"):
        raise ValueError('state must be "open", "closed", or "all".')
    limit = max(1, min(limit, 100))

    params: dict[str, Any] = {
        "state": state,
        "per_page": limit,
        "sort": "created",
        "direction": "desc",
    }

    raw: list[dict[str, Any]] = await _github_request(  # type: ignore[assignment]
        "GET", f"/repos/{owner}/{repo}/pulls", params=params
    )

    results = []
    for pr in raw[:limit]:
        results.append(
            {
                "number": pr["number"],
                "title": pr["title"],
                "state": pr["state"],
                "author": pr["user"]["login"],
                "created_at": pr["created_at"],
                "updated_at": pr["updated_at"],
                "head_branch": pr["head"]["ref"],
                "base_branch": pr["base"]["ref"],
                "draft": pr.get("draft", False),
                "mergeable": pr.get("mergeable"),
                "merged_at": pr.get("merged_at"),
                "html_url": pr["html_url"],
            }
        )

    return {
        "repository": f"{owner}/{repo}",
        "filter": {"state": state},
        "total_returned": len(results),
        "pull_requests": results,
        "_note": _rate_limit_note(),
    }


# ---------------------------------------------------------------------------
# Tool 4 — Contributor Stats
# ---------------------------------------------------------------------------


@mcp.tool()
async def get_contributor_stats(
    owner: str,
    repo: str,
    limit: int = 10,
) -> dict[str, Any]:
    """Get the top contributors to a GitHub repository, sorted by commit count.

    Args:
        owner: Repository owner (user or organization).
        repo:  Repository name.
        limit: Maximum number of contributors to return (1-100). Default: 10.
    """
    limit = max(1, min(limit, 100))

    params: dict[str, Any] = {"per_page": limit}
    raw: list[dict[str, Any]] = await _github_request(  # type: ignore[assignment]
        "GET", f"/repos/{owner}/{repo}/contributors", params=params
    )

    results = []
    for idx, contributor in enumerate(raw[:limit], start=1):
        results.append(
            {
                "rank": idx,
                "username": contributor["login"],
                "contributions": contributor["contributions"],
                "avatar_url": contributor["avatar_url"],
                "profile_url": contributor["html_url"],
            }
        )

    return {
        "repository": f"{owner}/{repo}",
        "total_returned": len(results),
        "contributors": results,
        "_note": _rate_limit_note(),
    }


# ---------------------------------------------------------------------------
# Tool 5 — Repository Search
# ---------------------------------------------------------------------------


@mcp.tool()
async def search_repositories(
    query: str,
    language: Optional[str] = None,
    sort: str = "stars",
    limit: int = 10,
) -> dict[str, Any]:
    """Search GitHub repositories by keyword, with optional language filter.

    Args:
        query:    Search query (e.g. "machine learning", "fastapi framework").
        language: Filter results by primary language (e.g. "python", "rust").
        sort:     Sort order — "stars", "forks", or "updated". Default: "stars".
        limit:    Maximum number of results to return (1-100). Default: 10.
    """
    if sort not in ("stars", "forks", "updated"):
        raise ValueError('sort must be "stars", "forks", or "updated".')
    limit = max(1, min(limit, 100))

    q = query
    if language:
        q += f" language:{language}"

    params: dict[str, Any] = {
        "q": q,
        "sort": sort,
        "order": "desc",
        "per_page": limit,
    }

    data: dict[str, Any] = await _github_request(  # type: ignore[assignment]
        "GET", "/search/repositories", params=params
    )

    results = []
    for item in data.get("items", [])[:limit]:
        license_info = item.get("license")
        results.append(
            {
                "full_name": item["full_name"],
                "description": item.get("description"),
                "stars": item["stargazers_count"],
                "forks": item["forks_count"],
                "language": item.get("language"),
                "license": license_info["spdx_id"] if license_info else None,
                "updated_at": item["updated_at"],
                "html_url": item["html_url"],
            }
        )

    return {
        "query": query,
        "language_filter": language,
        "sort_by": sort,
        "total_results_available": data.get("total_count", 0),
        "total_returned": len(results),
        "repositories": results,
        "_note": _rate_limit_note(),
    }


# ---------------------------------------------------------------------------
# Startup logging
# ---------------------------------------------------------------------------


def _log_startup() -> None:
    """Log authentication mode on startup."""
    if GITHUB_TOKEN:
        masked = GITHUB_TOKEN[:4] + "..." + GITHUB_TOKEN[-4:]
        logger.info("Authenticated mode enabled (token: %s).", masked)
        logger.info("Rate limit: 5,000 requests/hour.")
    else:
        logger.info("Running in unauthenticated mode.")
        logger.info(
            "Rate limit: 60 requests/hour. "
            "Set GITHUB_TOKEN in .env for higher limits."
        )


# ---------------------------------------------------------------------------
# Entrypoint
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    _log_startup()
    mcp.run()