#!/usr/bin/env python3
"""
MCP Git Analysis Server
Analyzes local git repositories and provides insights
"""
import asyncio
import os
from typing import Any, Optional
from datetime import datetime
from mcp.server import Server
from mcp.types import (
Resource,
Tool,
TextContent,
ImageContent,
EmbeddedResource,
)
from mcp.server.stdio import stdio_server
import git
# Initialize MCP server
app = Server("git-analysis-server")
# Tool implementations
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List all available tools."""
return [
Tool(
name="analyze_commits",
description="Get commit statistics for a repository. Can filter by count, author, or date range.",
inputSchema={
"type": "object",
"properties": {
"repo_path": {
"type": "string",
"description": "Path to the git repository (defaults to current directory)",
},
"count": {
"type": "number",
"description": "Number of recent commits to analyze",
},
"author": {
"type": "string",
"description": "Filter commits by author name or email",
},
"since": {
"type": "string",
"description": "Start date (ISO format: YYYY-MM-DD)",
},
"until": {
"type": "string",
"description": "End date (ISO format: YYYY-MM-DD)",
},
},
"required": [],
},
),
Tool(
name="find_hot_spots",
description="Find the most frequently changed files in the repository",
inputSchema={
"type": "object",
"properties": {
"repo_path": {
"type": "string",
"description": "Path to the git repository (defaults to current directory)",
},
"limit": {
"type": "number",
"description": "Number of hot spot files to return (default: 10)",
},
},
"required": [],
},
),
Tool(
name="get_contributors",
description="List all contributors with their statistics",
inputSchema={
"type": "object",
"properties": {
"repo_path": {
"type": "string",
"description": "Path to the git repository (defaults to current directory)",
},
},
"required": [],
},
),
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""Handle tool calls."""
if name == "analyze_commits":
return await analyze_commits(arguments)
elif name == "find_hot_spots":
return await find_hot_spots(arguments)
elif name == "get_contributors":
return await get_contributors(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
async def analyze_commits(arguments: dict) -> list[TextContent]:
"""Analyze commits with optional filters."""
repo_path = arguments.get("repo_path", ".")
count = arguments.get("count")
author = arguments.get("author")
since = arguments.get("since")
until = arguments.get("until")
try:
repo = git.Repo(repo_path)
# Build commit filter
kwargs = {}
if count:
kwargs["max_count"] = int(count)
if since:
kwargs["since"] = since
if until:
kwargs["until"] = until
if author:
kwargs["author"] = author
commits = list(repo.iter_commits(**kwargs))
# Build response
result = {
"total_commits": len(commits),
"repository": repo_path,
"commits": []
}
for commit in commits:
result["commits"].append({
"sha": commit.hexsha[:8],
"author": str(commit.author),
"date": commit.committed_datetime.isoformat(),
"message": commit.message.strip(),
"files_changed": len(commit.stats.files),
"insertions": commit.stats.total["insertions"],
"deletions": commit.stats.total["deletions"],
})
return [TextContent(type="text", text=str(result))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def find_hot_spots(arguments: dict) -> list[TextContent]:
"""Find most frequently changed files."""
repo_path = arguments.get("repo_path", ".")
limit = arguments.get("limit", 10)
try:
repo = git.Repo(repo_path)
# Track file changes and authors
file_stats = {}
for commit in repo.iter_commits():
for file_path in commit.stats.files.keys():
if file_path not in file_stats:
file_stats[file_path] = {
"changes": 0,
"authors": set()
}
file_stats[file_path]["changes"] += 1
file_stats[file_path]["authors"].add(str(commit.author))
# Sort by change count
sorted_files = sorted(
file_stats.items(),
key=lambda x: x[1]["changes"],
reverse=True
)[:int(limit)]
# Build response
result = {
"hot_spots": [
{
"file": file_path,
"changes": stats["changes"],
"authors": len(stats["authors"])
}
for file_path, stats in sorted_files
]
}
return [TextContent(type="text", text=str(result))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
async def get_contributors(arguments: dict) -> list[TextContent]:
"""Get all contributors with their statistics."""
repo_path = arguments.get("repo_path", ".")
try:
repo = git.Repo(repo_path)
# Track contributor stats
contributors = {}
for commit in repo.iter_commits():
author = str(commit.author)
if author not in contributors:
contributors[author] = {
"commits": 0,
"insertions": 0,
"deletions": 0,
"first_commit": commit.committed_datetime.isoformat(),
"last_commit": commit.committed_datetime.isoformat(),
}
contributors[author]["commits"] += 1
contributors[author]["insertions"] += commit.stats.total["insertions"]
contributors[author]["deletions"] += commit.stats.total["deletions"]
# Update dates
commit_date = commit.committed_datetime.isoformat()
if commit_date < contributors[author]["first_commit"]:
contributors[author]["first_commit"] = commit_date
if commit_date > contributors[author]["last_commit"]:
contributors[author]["last_commit"] = commit_date
# Sort by commit count
sorted_contributors = sorted(
contributors.items(),
key=lambda x: x[1]["commits"],
reverse=True
)
result = {
"total_contributors": len(contributors),
"contributors": [
{
"author": author,
**stats
}
for author, stats in sorted_contributors
]
}
return [TextContent(type="text", text=str(result))]
except Exception as e:
return [TextContent(type="text", text=f"Error: {str(e)}")]
# Resource implementations
@app.list_resources()
async def list_resources() -> list[Resource]:
"""List available resources."""
return [
Resource(
uri="git://recent-activity",
name="Recent Git Activity",
mimeType="application/json",
description="Last 10 commits as structured data",
),
Resource(
uri="git://file-history/{filepath}",
name="File History",
mimeType="application/json",
description="Change history for a specific file",
),
]
@app.read_resource()
async def read_resource(uri: str) -> str:
"""Handle resource requests."""
# Convert uri to string if it's an AnyUrl object
uri_str = str(uri)
if uri_str == "git://recent-activity":
return await get_recent_activity()
elif uri_str.startswith("git://file-history/"):
filepath = uri_str.replace("git://file-history/", "")
return await get_file_history(filepath)
else:
raise ValueError(f"Unknown resource: {uri_str}")
async def get_recent_activity() -> str:
"""Get last 10 commits as structured data."""
try:
repo = git.Repo(".")
commits = list(repo.iter_commits(max_count=10))
result = {
"recent_activity": [
{
"sha": commit.hexsha[:8],
"author": str(commit.author),
"date": commit.committed_datetime.isoformat(),
"message": commit.message.strip(),
"files_changed": len(commit.stats.files),
}
for commit in commits
]
}
return str(result)
except Exception as e:
return f"Error: {str(e)}"
async def get_file_history(filepath: str) -> str:
"""Get change history for a specific file."""
try:
repo = git.Repo(".")
commits = list(repo.iter_commits(paths=filepath))
result = {
"file": filepath,
"total_commits": len(commits),
"history": [
{
"sha": commit.hexsha[:8],
"author": str(commit.author),
"date": commit.committed_datetime.isoformat(),
"message": commit.message.strip(),
}
for commit in commits[:20] # Limit to 20 most recent
]
}
return str(result)
except Exception as e:
return f"Error: {str(e)}"
async def main():
"""Run the MCP server."""
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
