server.py

import os
import json
import base64
import httpx
from typing import List, Optional, Dict, Any
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

# Initialize FastMCP server
mcp = FastMCP("GitHub")

# Load environment variables
load_dotenv()

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

def get_headers():
    headers = {
        "Accept": "application/vnd.github.v3+json",
    }
    if GITHUB_TOKEN:
        headers["Authorization"] = f"token {GITHUB_TOKEN}"
    return headers

@mcp.tool()
async def list_repositories() -> List[Dict[str, Any]]:
    """
    Lists repositories for the authenticated user.
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(f"{BASE_URL}/user/repos", headers=get_headers())
        if response.status_code != 200:
            return [{"error": f"Failed to list repos: {response.text}"}]
        return response.json()

@mcp.tool()
async def create_repository(name: str, description: str = "", private: bool = False) -> Dict[str, Any]:
    """
    Creates a new repository for the authenticated user.
    """
    async with httpx.AsyncClient() as client:
        data = {"name": name, "description": description, "private": private}
        response = await client.post(f"{BASE_URL}/user/repos", headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def get_file_contents(owner: str, repo: str, path: str) -> Dict[str, Any]:
    """
    Retrieves the content and metadata of a file in a repository.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/contents/{path}"
        response = await client.get(url, headers=get_headers())
        if response.status_code != 200:
            return {"error": response.text}
        
        data = response.json()
        if "content" in data and data.get("encoding") == "base64":
            data["decoded_content"] = base64.b64decode(data["content"]).decode("utf-8")
        return data

@mcp.tool()
async def create_or_update_file(owner: str, repo: str, path: str, content: str, message: str, branch: str = "main", sha: Optional[str] = None) -> Dict[str, Any]:
    """
    Creates or updates a file in a repository. 'sha' is required for updates.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/contents/{path}"
        encoded_content = base64.b64encode(content.encode("utf-8")).decode("utf-8")
        data = {
            "message": message,
            "content": encoded_content,
            "branch": branch
        }
        if sha:
            data["sha"] = sha
            
        response = await client.put(url, headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def list_issues(owner: str, repo: str, state: str = "open") -> List[Dict[str, Any]]:
    """
    Lists issues for a specific repository.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/issues"
        params = {"state": state}
        response = await client.get(url, headers=get_headers(), params=params)
        return response.json()

@mcp.tool()
async def create_issue(owner: str, repo: str, title: str, body: str = "") -> Dict[str, Any]:
    """
    Creates a new issue in a repository.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/issues"
        data = {"title": title, "body": body}
        response = await client.post(url, headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def list_pull_requests(owner: str, repo: str, state: str = "open") -> List[Dict[str, Any]]:
    """
    Lists pull requests for a repository.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/pulls"
        params = {"state": state}
        response = await client.get(url, headers=get_headers(), params=params)
        return response.json()

@mcp.tool()
async def search_repositories(query: str) -> Dict[str, Any]:
    """
    Searches GitHub for repositories.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/search/repositories"
        params = {"q": query}
        response = await client.get(url, headers=get_headers(), params=params)
        return response.json()

@mcp.tool()
async def create_branch(owner: str, repo: str, branch_name: str, base_branch: str = "main") -> Dict[str, Any]:
    """
    Creates a new branch from a base branch.
    """
    async with httpx.AsyncClient() as client:
        # Get base branch SHA
        ref_res = await client.get(f"{BASE_URL}/repos/{owner}/{repo}/git/ref/heads/{base_branch}", headers=get_headers())
        if ref_res.status_code != 200:
            return {"error": f"Base branch not found: {ref_res.text}"}
        
        sha = ref_res.json()["object"]["sha"]
        
        # Create new ref
        data = {"ref": f"refs/heads/{branch_name}", "sha": sha}
        response = await client.post(f"{BASE_URL}/repos/{owner}/{repo}/git/refs", headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def merge_pull_request(owner: str, repo: str, pr_number: int) -> Dict[str, Any]:
    """
    Merges an open pull request.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/pulls/{pr_number}/merge"
        response = await client.put(url, headers=get_headers())
        return response.json()

@mcp.tool()
async def list_commits(owner: str, repo: str, branch: str = "main") -> List[Dict[str, Any]]:
    """
    Lists recent commits for a repository and branch.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/commits"
        params = {"sha": branch}
        response = await client.get(url, headers=get_headers(), params=params)
        return response.json()

@mcp.tool()
async def create_release(owner: str, repo: str, tag_name: str, name: str, body: str = "") -> Dict[str, Any]:
    """
    Publishes a new GitHub release.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/releases"
        data = {"tag_name": tag_name, "name": name, "body": body}
        response = await client.post(url, headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def add_collaborator(owner: str, repo: str, username: str, permission: str = "push") -> Dict[str, Any]:
    """
    Invites a collaborator to a repository. Permission: pull, push, admin, maintain, triage.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/collaborators/{username}"
        data = {"permission": permission}
        response = await client.put(url, headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def list_workflow_runs(owner: str, repo: str) -> Dict[str, Any]:
    """
    Lists recent GitHub Actions workflow runs for a repository.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/repos/{owner}/{repo}/actions/runs"
        response = await client.get(url, headers=get_headers())
        return response.json()

@mcp.tool()
async def create_gist(description: str, files: Dict[str, Dict[str, str]], public: bool = False) -> Dict[str, Any]:
    """
    Creates a new Gist. 'files' is a dict like {'filename.txt': {'content': 'hello world'}}.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/gists"
        data = {"description": description, "files": files, "public": public}
        response = await client.post(url, headers=get_headers(), json=data)
        return response.json()

@mcp.tool()
async def get_user_info(username: str) -> Dict[str, Any]:
    """
    Retrieves profile data for a specific GitHub user.
    """
    async with httpx.AsyncClient() as client:
        url = f"{BASE_URL}/users/{username}"
        response = await client.get(url, headers=get_headers())
        return response.json()

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