server.py

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

load_dotenv()

GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
GITHUB_API_BASE = "https://api.github.com"

mcp = FastMCP("github-mcp-server")

@mcp.tool()
async def list_repos() -> str:
    """List all repositories (public and private) for the authenticated GitHub user."""
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{GITHUB_API_BASE}/user/repos",
            headers=headers,
            params={"per_page": 100, "visibility": "all"}
        )
        response.raise_for_status()
        repos = response.json()

    if not repos:
        return "No repositories found."
    lines = []
    for repo in repos:
        visibility = "private" if repo["private"] else "public"
        lines.append(f"- {repo['name']} ({visibility})")
    return "\n".join(lines)

@mcp.tool()
async def get_issues(owner: str, repo_name: str) -> str:
    """Get issues (open and closed) for a specific GitHub repository.

    Args:
        owner: The GitHub username or organization that owns the repository (e.g. "Lilli-Ai").
        repo_name: The name of the repository (e.g. "weather-mcp-server").
    """
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json"
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"{GITHUB_API_BASE}/repos/{owner}/{repo_name}/issues",
            headers=headers,
            params={"state": "all", "per_page": 100}
        )
        response.raise_for_status()
        issues = response.json()

    if not issues:
        return f"No issues found in {owner}/{repo_name}."

    lines = []
    for issue in issues:
        item_type = "PR" if "pull_request" in issue else "issue"
        lines.append(f"#{issue['number']} [{issue['state']}] ({item_type}): {issue['title']}")
    return "\n".join(lines)

@mcp.tool()
async def get_all_issues() -> str:
    """Get issues (open and closed) across all repositories the authenticated user has access to."""
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json"
    }
    async with httpx.AsyncClient() as client:
        repos_response = await client.get(
            f"{GITHUB_API_BASE}/user/repos",
            headers=headers,
            params={"per_page": 100, "visibility": "all"}
        )
        repos_response.raise_for_status()
        repos = repos_response.json()

        result_lines = []
        for repo in repos:
            owner = repo["owner"]["login"]
            name = repo["name"]
            issues_response = await client.get(
                f"{GITHUB_API_BASE}/repos/{owner}/{name}/issues",
                headers=headers,
                params={"state": "all", "per_page": 100}
            )
            issues_response.raise_for_status()
            issues = issues_response.json()

            if issues:
                result_lines.append(f"\n{owner}/{name}:")
                for issue in issues:
                    item_type = "PR" if "pull_request" in issue else "issue"
                    result_lines.append(f"  #{issue['number']} [{issue['state']}] ({item_type}): {issue['title']}")

    if not result_lines:
        return "No issues found across any repository."

    return "\n".join(result_lines)

@mcp.tool()
async def create_issue(owner: str, repo_name: str, title: str, body: str = "") -> str:
    """Create a new issue in a specific GitHub repository.

    Args:
        owner: The GitHub username or organization that owns the repository (e.g. "Lilli-Ai").
        repo_name: The name of the repository (e.g. "github-mcp-demo").
        title: The title of the issue.
        body: Optional description/body text for the issue.
    """
    headers = {
        "Authorization": f"Bearer {GITHUB_TOKEN}",
        "Accept": "application/vnd.github+json"
    }
    payload = {"title": title, "body": body}

    async with httpx.AsyncClient() as client:
        response = await client.post(
            f"{GITHUB_API_BASE}/repos/{owner}/{repo_name}/issues",
            headers=headers,
            json=payload
        )
        response.raise_for_status()
        issue = response.json()

    return f"Created issue #{issue['number']}: {issue['title']}\n{issue['html_url']}"

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