src/mcp-github/server.py

"""
GitHub MCP Server
"""
import sys
import os
from mcp.server.fastmcp import FastMCP
from typing import List, Dict, Any, Optional
from core.mcp_telemetry import log_usage, log_trace, log_metric
import uuid
import time
import datetime

# Add src to pythonpath
current_dir = os.path.dirname(os.path.abspath(__file__))
src_dir = os.path.dirname(os.path.dirname(current_dir))
if src_dir not in sys.path:
    sys.path.append(src_dir)

try:
    from github import Github, Auth
    from github.GithubException import GithubException
except ImportError:
    Github = None
    Auth = None
    GithubException = Exception

# Initialize FastMCP Server
mcp = FastMCP("GitHub Operations", host="0.0.0.0")

def get_client():
    token = os.environ.get("GITHUB_TOKEN")
    if not token:
        raise ValueError("GITHUB_TOKEN environment variable not set.")
    if not Github:
        raise ImportError("PyGithub not installed.")
    auth = Auth.Token(token)
    return Github(auth=auth)

@mcp.tool()
def list_repositories() -> List[Dict[str, Any]]:
    """
    List all repositories for the authenticated user/owner.
    """
    start_time = time.time()
    trace_id = str(uuid.uuid4())
    span_id = str(uuid.uuid4())
    log_usage("mcp-github", "list_repositories")
    
    try:
        g = get_client()
        # Get repos for the owner/authenticated user
        repos = g.get_user().get_repos(sort="updated", direction="desc")
        
        results = []
        for repo in repos[:20]: # Limit to 20 most recent
            results.append({
                "name": repo.name,
                "full_name": repo.full_name,
                "description": repo.description,
                "stars": repo.stargazers_count,
                "forks": repo.forks_count,
                "updated_at": str(repo.updated_at),
                "language": repo.language
            })
        
        duration = (time.time() - start_time) * 1000
        log_trace("mcp-github", trace_id, span_id, "list_repositories", duration, "ok")
        log_metric("mcp-github", "repos_fetched", len(results), {"status": "ok"})
        return results
        
    except Exception as e:
        duration = (time.time() - start_time) * 1000
        log_trace("mcp-github", trace_id, span_id, "list_repositories", duration, "error")
        return [{"error": str(e)}]

@mcp.tool()
def list_issues(owner: str, repo_name: str, state: str = "open") -> List[Dict[str, Any]]:
    """
    List issues for a repository.
    """
    log_usage("mcp-github", "list_issues")
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        issues = repo.get_issues(state=state)
        
        results = []
        for issue in issues[:10]: # Limit to 10 recent
            results.append({
                "number": issue.number,
                "title": issue.title,
                "state": issue.state,
                "created_at": str(issue.created_at),
                "user": issue.user.login
            })
        return results
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
def create_issue(owner: str, repo_name: str, title: str, body: str) -> Dict[str, Any]:
    """
    Create a new issue.
    """
    log_usage("mcp-github", "create_issue")
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        issue = repo.create_issue(title=title, body=body)
        return {
            "number": issue.number,
            "title": issue.title,
            "url": issue.html_url
        }
    except Exception as e:
        return {"error": str(e)}

@mcp.tool()
def get_issue(owner: str, repo_name: str, issue_number: int) -> Dict[str, Any]:
    """
    Get detailed issue info including comments.
    """
    log_usage("mcp-github", "get_issue")
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        issue = repo.get_issue(issue_number)
        
        comments = []
        for c in issue.get_comments():
            comments.append({
                "user": c.user.login,
                "body": c.body,
                "created_at": str(c.created_at)
            })
            
        return {
            "number": issue.number,
            "title": issue.title,
            "body": issue.body,
            "state": issue.state,
            "comments": comments
        }
    except Exception as e:
        return {"error": str(e)}

@mcp.tool()
def list_security_alerts(owner: str, repo_name: str) -> List[Dict[str, Any]]:
    """
    List dependabot alerts (if enabled and accessible).
    """
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        
        # PyGithub support for alerts varies, using common method if available
        # or manual requests otherwise. Assuming PyGithub >= 2.0 has some support.
        # Often requires specific permissions.
        try:
            alerts = repo.get_dependabot_alerts()
            results = []
            for alert in alerts:
                results.append({
                    "number": alert.number,
                    "package": alert.dependency.package.name,
                    "severity": alert.security_advisory.severity,
                    "state": alert.state,
                    "created_at": str(alert.created_at)
                })
            return results
        except AttributeError:
             return [{"error": "get_dependabot_alerts not supported by this PyGithub version"}]
             
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
def create_pull_request(owner: str, repo_name: str, title: str, body: str, head: str, base: str = "main") -> Dict[str, Any]:
    """
    Create a new Pull Request.
    """
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        pr = repo.create_pull(title=title, body=body, head=head, base=base)
        return {
            "number": pr.number,
            "title": pr.title,
            "url": pr.html_url,
            "state": pr.state
        }
    except Exception as e:
        return {"error": str(e)}

@mcp.tool()
def get_pull_request(owner: str, repo_name: str, pr_number: int) -> Dict[str, Any]:
    """
    Get Pull Request details.
    """
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        pr = repo.get_pull(pr_number)
        return {
            "number": pr.number,
            "title": pr.title,
            "body": pr.body,
            "state": pr.state,
            "merged": pr.merged,
             "url": pr.html_url
        }
    except Exception as e:
         return {"error": str(e)}

@mcp.tool()
def list_workflow_runs(owner: str, repo_name: str) -> List[Dict[str, Any]]:
    """
    List recent workflow runs for a repository.
    """
    log_usage("mcp-github", "list_workflow_runs")
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        runs = repo.get_workflow_runs()
        
        results = []
        for run in runs[:10]:
            results.append({
                "id": run.id,
                "name": run.name,
                "status": run.status,
                "conclusion": run.conclusion,
                "event": run.event,
                "created_at": str(run.created_at),
                "url": run.html_url
            })
        return results
    except Exception as e:
        return [{"error": str(e)}]

@mcp.tool()
def get_workflow_run_details(owner: str, repo_name: str, run_id: int) -> Dict[str, Any]:
    """
    Get details of a specific workflow run.
    """
    log_usage("mcp-github", "get_workflow_run_details")
    try:
        g = get_client()
        repo = g.get_repo(f"{owner}/{repo_name}")
        run = repo.get_workflow_run(run_id)
        
        return {
            "id": run.id,
            "name": run.name,
            "status": run.status,
            "conclusion": run.conclusion,
            "event": run.event,
            "created_at": str(run.created_at),
            "updated_at": str(run.updated_at),
            "url": run.html_url,
            "run_number": run.run_number
        }
    except Exception as e:
        return {"error": str(e)}

if __name__ == "__main__":
    import os
    if os.environ.get("MCP_TRANSPORT") == "sse":
        import uvicorn
        port = int(os.environ.get("PORT", 7860))
        uvicorn.run(mcp.sse_app(), host="0.0.0.0", port=port)
    else:
        mcp.run()