server.py•12.8 kB
"""Main HTTP MCP server using FastAPI."""
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import json
from typing import Any, Dict
import uvicorn
from mcp.server import Server
from mcp.types import Tool, TextContent
# Import our tools
import config
from git_tools import (
git_status,
git_log,
git_diff,
git_list_branches,
git_stage_files,
git_commit,
git_checkout_branch
)
from github_tools import (
github_list_prs,
github_create_pr,
github_comment_on_pr
)
# Create FastAPI app
app = FastAPI(title="git-workflow-mcp", version="0.1.0")
# Create MCP server instance
mcp_server = Server("git-workflow-mcp")
# Register Git tools
@mcp_server.list_tools()
async def list_tools() -> list[Tool]:
"""List all available tools."""
tools = [
Tool(
name="git_status",
description="Get the current Git status including branch, ahead/behind counts, and lists of staged, modified, and untracked files.",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="git_log",
description="Get Git commit log with author, date, and message. Returns recent commits up to a specified limit.",
inputSchema={
"type": "object",
"properties": {
"limit": {
"type": "integer",
"description": "Number of commits to return (1-50)",
"minimum": 1,
"maximum": 50,
"default": 10
}
},
"required": []
}
),
Tool(
name="git_diff",
description="Get the diff for a specific file showing changes.",
inputSchema={
"type": "object",
"properties": {
"file_path": {
"type": "string",
"description": "Relative path to the file"
}
},
"required": ["file_path"]
}
),
Tool(
name="git_list_branches",
description="List all Git branches and show the current branch.",
inputSchema={
"type": "object",
"properties": {},
"required": []
}
),
Tool(
name="git_stage_files",
description="Stage files for commit. Validates that paths are within the repository.",
inputSchema={
"type": "object",
"properties": {
"files": {
"type": "array",
"items": {"type": "string"},
"description": "List of relative file paths to stage"
}
},
"required": ["files"]
}
),
Tool(
name="git_commit",
description="Create a commit with the given message. Requires staged changes.",
inputSchema={
"type": "object",
"properties": {
"message": {
"type": "string",
"description": "Commit message (minimum 3 characters)"
}
},
"required": ["message"]
}
),
Tool(
name="git_checkout_branch",
description="Checkout a branch, optionally creating it if it doesn't exist.",
inputSchema={
"type": "object",
"properties": {
"branch": {
"type": "string",
"description": "Branch name to checkout"
},
"create": {
"type": "boolean",
"description": "If true, create the branch if it doesn't exist",
"default": False
}
},
"required": ["branch"]
}
),
]
# Add GitHub tools if token is set
if config.GITHUB_TOKEN:
tools.extend([
Tool(
name="github_list_prs",
description="List pull requests for the repository. Requires GITHUB_TOKEN.",
inputSchema={
"type": "object",
"properties": {
"state": {
"type": "string",
"enum": ["open", "closed", "all"],
"description": "PR state filter",
"default": "open"
}
},
"required": []
}
),
Tool(
name="github_create_pr",
description="Create a new pull request. Requires GITHUB_TOKEN.",
inputSchema={
"type": "object",
"properties": {
"title": {
"type": "string",
"description": "PR title"
},
"body": {
"type": "string",
"description": "PR description (optional)",
"default": ""
},
"head": {
"type": "string",
"description": "Source branch (defaults to current branch)",
"default": ""
},
"base": {
"type": "string",
"description": "Target branch",
"default": "main"
}
},
"required": ["title"]
}
),
Tool(
name="github_comment_on_pr",
description="Add a comment to a pull request. Requires GITHUB_TOKEN.",
inputSchema={
"type": "object",
"properties": {
"pr_number": {
"type": "integer",
"description": "PR number"
},
"body": {
"type": "string",
"description": "Comment text"
}
},
"required": ["pr_number", "body"]
}
),
])
return tools
@mcp_server.call_tool()
async def call_tool(name: str, arguments: Dict[str, Any]) -> list[TextContent]:
"""Handle tool calls."""
try:
if name == "git_status":
result = git_status()
elif name == "git_log":
limit = arguments.get("limit", 10)
result = git_log(limit=limit)
elif name == "git_diff":
file_path = arguments.get("file_path")
if not file_path:
raise ValueError("file_path is required")
result = git_diff(file_path)
elif name == "git_list_branches":
result = git_list_branches()
elif name == "git_stage_files":
files = arguments.get("files", [])
if not files:
raise ValueError("files list is required")
result = git_stage_files(files)
elif name == "git_commit":
message = arguments.get("message")
if not message:
raise ValueError("message is required")
result = git_commit(message)
elif name == "git_checkout_branch":
branch = arguments.get("branch")
if not branch:
raise ValueError("branch is required")
create = arguments.get("create", False)
result = git_checkout_branch(branch, create=create)
elif name == "github_list_prs":
state = arguments.get("state", "open")
result = github_list_prs(state=state)
elif name == "github_create_pr":
title = arguments.get("title")
if not title:
raise ValueError("title is required")
body = arguments.get("body", "")
head = arguments.get("head", "")
base = arguments.get("base", "main")
result = github_create_pr(title, body, head, base)
elif name == "github_comment_on_pr":
pr_number = arguments.get("pr_number")
body = arguments.get("body")
if pr_number is None:
raise ValueError("pr_number is required")
if not body:
raise ValueError("body is required")
result = github_comment_on_pr(pr_number, body)
else:
raise ValueError(f"Unknown tool: {name}")
# Format result as JSON string for content
content_text = result.get("content", json.dumps(result, indent=2))
return [TextContent(
type="text",
text=content_text
)]
except Exception as e:
error_msg = f"Error executing {name}: {str(e)}"
return [TextContent(
type="text",
text=error_msg
)]
@app.post("/mcp")
async def mcp_endpoint(request: Request):
"""Handle MCP requests via HTTP."""
try:
body = await request.json()
# Handle MCP protocol messages
if body.get("jsonrpc") != "2.0":
raise HTTPException(status_code=400, detail="Invalid JSON-RPC version")
method = body.get("method")
params = body.get("params", {})
request_id = body.get("id")
if method == "tools/list":
tools = await list_tools()
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"tools": [
{
"name": tool.name,
"description": tool.description,
"inputSchema": tool.inputSchema
}
for tool in tools
]
}
}
elif method == "tools/call":
tool_name = params.get("name")
arguments = params.get("arguments", {})
if not tool_name:
raise HTTPException(status_code=400, detail="Tool name is required")
contents = await call_tool(tool_name, arguments)
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"content": [
{
"type": content.type,
"text": content.text
}
for content in contents
]
}
}
elif method == "initialize":
return {
"jsonrpc": "2.0",
"id": request_id,
"result": {
"protocolVersion": "2024-11-05",
"capabilities": {
"tools": {}
},
"serverInfo": {
"name": "git-workflow-mcp",
"version": "0.1.0"
}
}
}
else:
raise HTTPException(status_code=400, detail=f"Unknown method: {method}")
except HTTPException:
raise
except Exception as e:
return JSONResponse(
status_code=500,
content={
"jsonrpc": "2.0",
"id": body.get("id") if 'body' in locals() else None,
"error": {
"code": -32603,
"message": str(e)
}
}
)
@app.get("/")
async def root():
"""Health check endpoint."""
return {
"name": "git-workflow-mcp",
"version": "0.1.0",
"status": "running",
"github_enabled": config.GITHUB_TOKEN is not None
}
if __name__ == "__main__":
import os
host = os.environ.get("MCP_HOST", "127.0.0.1")
port = int(os.environ.get("MCP_PORT", "3333"))
uvicorn.run(app, host=host, port=port)