server.py•10.3 kB
#!/usr/bin/env python3
"""
GitHub MCP Server
A FastMCP server that provides access to GitHub repositories and data.
"""
import asyncio
import json
import os
from typing import Any, Dict, List, Optional
import httpx
from fastmcp import FastMCP
# Create the FastMCP server
mcp = FastMCP("GitHub MCP Server")
# GitHub API configuration
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN") or os.getenv("GITHUB_API_KEY") or "demo"
GITHUB_API_BASE = "https://api.github.com"
def get_headers(api_key=None):
"""Get headers for GitHub API requests."""
headers = {
"Accept": "application/vnd.github.v3+json",
"User-Agent": "GitHub-MCP-Server"
}
# Use provided API key or fallback to environment variable
token = api_key or GITHUB_TOKEN
if token and token != "demo":
headers["Authorization"] = f"Bearer {token}"
return headers
@mcp.tool()
def get_repos(username: str = None, limit: int = 10) -> str:
"""Get repositories for a GitHub user.
Args:
username: GitHub username (optional); if not provided, returns authenticated user's repos
limit: Number of repositories to return (default: 10, max: 30)
Returns:
JSON string with repository data
"""
try:
limit = min(max(limit, 1), 30) # Clamp between 1 and 30
token = GITHUB_TOKEN
if not token or token == "demo":
return json.dumps({"error": "GitHub token not configured. Please set GITHUB_TOKEN environment variable."}, indent=2)
params = {
"sort": "updated",
"per_page": limit
}
# Determine endpoint based on username parameter
if username and username.strip():
# Default to public-only endpoint for arbitrary users
endpoint = f"{GITHUB_API_BASE}/users/{username}/repos"
# If the provided username matches the authenticated user, use /user/repos to include private repos
try:
me_response = httpx.get(
f"{GITHUB_API_BASE}/user",
headers=get_headers(token),
timeout=10.0
)
me_response.raise_for_status()
me = me_response.json()
auth_login = (me.get("login") or "").strip()
if auth_login and auth_login.lower() == username.strip().lower():
endpoint = f"{GITHUB_API_BASE}/user/repos"
except httpx.RequestError:
# Fall back to public endpoint if we cannot determine the authenticated user
pass
except httpx.HTTPStatusError:
# Fall back to public endpoint if token is invalid/insufficient
pass
else:
# Get repos for authenticated user (public + private)
# This happens when username is None or empty string
endpoint = f"{GITHUB_API_BASE}/user/repos"
response = httpx.get(
endpoint,
headers=get_headers(token),
params=params,
timeout=10.0
)
response.raise_for_status()
repos = response.json()
formatted_repos = []
for repo in repos:
formatted_repos.append({
"name": repo["name"],
"full_name": repo["full_name"],
"description": repo.get("description", ""),
"url": repo["html_url"],
"stars": repo["stargazers_count"],
"forks": repo["forks_count"],
"language": repo.get("language", ""),
"created_at": repo["created_at"],
"updated_at": repo["updated_at"],
"private": repo["private"]
})
return json.dumps(formatted_repos, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def get_issues(owner: str, repo: str, state: str = "open", limit: int = 10) -> str:
"""Get issues for a GitHub repository.
Args:
owner: Repository owner
repo: Repository name
state: Issue state: open, closed, or all (default: open)
limit: Number of issues to return (default: 10, max: 30)
Returns:
JSON string with issues data
"""
try:
limit = min(max(limit, 1), 30) # Clamp between 1 and 30
token = GITHUB_TOKEN
if not token or token == "demo":
return json.dumps({"error": "GitHub token not configured. Please set GITHUB_TOKEN environment variable."}, indent=2)
params = {
"state": state,
"per_page": limit,
"sort": "updated"
}
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/issues",
headers=get_headers(token),
params=params,
timeout=10.0
)
response.raise_for_status()
issues = response.json()
formatted_issues = []
for issue in issues:
# Skip pull requests (they appear in issues API)
if "pull_request" in issue:
continue
formatted_issues.append({
"number": issue["number"],
"title": issue["title"],
"body": issue.get("body", ""),
"state": issue["state"],
"url": issue["html_url"],
"user": issue["user"]["login"],
"labels": [label["name"] for label in issue["labels"]],
"created_at": issue["created_at"],
"updated_at": issue["updated_at"],
"comments": issue["comments"]
})
return json.dumps(formatted_issues, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def get_pull_requests(owner: str, repo: str, state: str = "open", limit: int = 10) -> str:
"""Get pull requests for a GitHub repository.
Args:
owner: Repository owner
repo: Repository name
state: PR state: open, closed, or all (default: open)
limit: Number of PRs to return (default: 10, max: 30)
Returns:
JSON string with pull requests data
"""
try:
limit = min(max(limit, 1), 30) # Clamp between 1 and 30
token = GITHUB_TOKEN
if not token or token == "demo":
return json.dumps({"error": "GitHub token not configured. Please set GITHUB_TOKEN environment variable."}, indent=2)
params = {
"state": state,
"per_page": limit,
"sort": "updated"
}
response = httpx.get(
f"{GITHUB_API_BASE}/repos/{owner}/{repo}/pulls",
headers=get_headers(token),
params=params,
timeout=10.0
)
response.raise_for_status()
prs = response.json()
formatted_prs = []
for pr in prs:
formatted_prs.append({
"number": pr["number"],
"title": pr["title"],
"body": pr.get("body", ""),
"state": pr["state"],
"url": pr["html_url"],
"user": pr["user"]["login"],
"head": pr["head"]["ref"],
"base": pr["base"]["ref"],
"created_at": pr["created_at"],
"updated_at": pr["updated_at"],
"draft": pr["draft"],
"mergeable": pr.get("mergeable")
})
return json.dumps(formatted_prs, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
@mcp.tool()
def search_code(query: str, language: str = "", limit: int = 10) -> str:
"""Search for code on GitHub.
Args:
query: Search query
language: Programming language filter (optional)
limit: Number of results to return (default: 10, max: 20)
Returns:
JSON string with search results
"""
try:
limit = min(max(limit, 1), 20) # Clamp between 1 and 20
token = GITHUB_TOKEN
if not token or token == "demo":
return json.dumps({"error": "GitHub token not configured. Please set GITHUB_TOKEN environment variable."}, indent=2)
search_query = query
if language:
search_query += f" language:{language}"
params = {
"q": search_query,
"per_page": limit,
"sort": "indexed"
}
response = httpx.get(
f"{GITHUB_API_BASE}/search/code",
headers=get_headers(token),
params=params,
timeout=10.0
)
response.raise_for_status()
data = response.json()
results = []
for item in data.get("items", []):
results.append({
"name": item["name"],
"path": item["path"],
"repository": item["repository"]["full_name"],
"url": item["html_url"],
"language": item.get("language", ""),
"score": item["score"]
})
search_results = {
"total_count": data["total_count"],
"results": results
}
return json.dumps(search_results, indent=2)
except httpx.RequestError as e:
return json.dumps({"error": f"Request failed: {str(e)}"}, indent=2)
except Exception as e:
return json.dumps({"error": f"Unexpected error: {str(e)}"}, indent=2)
if __name__ == "__main__":
port = int(os.environ.get("PORT", 8000))
mcp.run(
transport="http",
host="0.0.0.0",
port=port,
stateless_http=True
)