github_tools.py•7.08 kB
"""GitHub API operations using requests."""
import subprocess
import re
from typing import Dict, Any, List, Optional
import requests
import config
def _get_github_repo_info() -> tuple[str, str]:
"""
Parse the origin remote URL to extract owner and repo name.
Returns:
Tuple of (owner, repo_name)
"""
if not config.GITHUB_TOKEN:
raise ValueError("GitHub integration is disabled because GITHUB_TOKEN is not set.")
try:
result = subprocess.run(
["git", "remote", "get-url", "origin"],
cwd=config.REPO_ROOT,
text=True,
capture_output=True,
check=True
)
url = result.stdout.strip()
# Parse GitHub URL (supports both https and ssh formats)
# https://github.com/owner/repo.git
# git@github.com:owner/repo.git
match = re.search(r'github\.com[:/]([^/]+)/([^/]+?)(?:\.git)?$', url)
if match:
owner = match.group(1)
repo = match.group(2)
return owner, repo
else:
raise ValueError(f"Could not parse GitHub URL from origin remote: {url}")
except subprocess.CalledProcessError:
raise ValueError("Could not get origin remote URL. Is this repository connected to GitHub?")
def _get_headers() -> Dict[str, str]:
"""Get headers for GitHub API requests."""
if not config.GITHUB_TOKEN:
raise ValueError("GitHub integration is disabled because GITHUB_TOKEN is not set.")
return {
"Authorization": f"Bearer {config.GITHUB_TOKEN}",
"Accept": "application/vnd.github+json",
"User-Agent": "git-workflow-mcp/0.1.0"
}
def github_list_prs(state: str = "open") -> Dict[str, Any]:
"""
List pull requests for the repository.
Args:
state: PR state - "open", "closed", or "all" (default: "open")
Returns:
Dictionary with list of pull requests.
"""
if not config.GITHUB_TOKEN:
return {
"error": "GitHub integration is disabled because GITHUB_TOKEN is not set.",
"content": "GitHub integration is disabled because GITHUB_TOKEN is not set."
}
if state not in ["open", "closed", "all"]:
raise ValueError("state must be one of: 'open', 'closed', 'all'")
owner, repo = _get_github_repo_info()
headers = _get_headers()
url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
params = {"state": state}
try:
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
prs_data = response.json()
pull_requests = []
for pr in prs_data:
pull_requests.append({
"number": pr["number"],
"title": pr["title"],
"user": pr["user"]["login"],
"state": pr["state"],
"html_url": pr["html_url"]
})
# Create human-readable summary
content_parts = [f"Pull Requests ({state}, {len(pull_requests)} found):"]
for pr in pull_requests:
content_parts.append(f"\n#{pr['number']}: {pr['title']}")
content_parts.append(f" State: {pr['state']}, Author: {pr['user']}")
content_parts.append(f" URL: {pr['html_url']}")
return {
"pull_requests": pull_requests,
"content": "\n".join(content_parts)
}
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Failed to fetch pull requests: {str(e)}")
def github_create_pr(title: str, body: str = "", head: str = "", base: str = "main") -> Dict[str, Any]:
"""
Create a new pull request.
Args:
title: PR title
body: PR description (optional)
head: Source branch (defaults to current branch)
base: Target branch (default: "main")
Returns:
Dictionary with PR number, title, state, and URL.
"""
if not config.GITHUB_TOKEN:
return {
"error": "GitHub integration is disabled because GITHUB_TOKEN is not set.",
"content": "GitHub integration is disabled because GITHUB_TOKEN is not set."
}
if not title or not title.strip():
raise ValueError("PR title cannot be empty")
owner, repo = _get_github_repo_info()
headers = _get_headers()
# If head is not provided, use current branch
if not head:
try:
result = subprocess.run(
["git", "rev-parse", "--abbrev-ref", "HEAD"],
cwd=config.REPO_ROOT,
text=True,
capture_output=True,
check=True
)
head = result.stdout.strip()
except subprocess.CalledProcessError:
raise ValueError("Could not determine current branch. Please specify 'head' parameter.")
url = f"https://api.github.com/repos/{owner}/{repo}/pulls"
data = {
"title": title,
"head": head,
"base": base
}
if body:
data["body"] = body
try:
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
pr_data = response.json()
result = {
"number": pr_data["number"],
"title": pr_data["title"],
"state": pr_data["state"],
"html_url": pr_data["html_url"]
}
result["content"] = f"Successfully created PR #{result['number']}: {result['title']}\nURL: {result['html_url']}"
return result
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Failed to create pull request: {str(e)}")
def github_comment_on_pr(pr_number: int, body: str) -> Dict[str, Any]:
"""
Add a comment to a pull request.
Args:
pr_number: PR number
body: Comment text
Returns:
Dictionary with success status and comment URL.
"""
if not config.GITHUB_TOKEN:
return {
"error": "GitHub integration is disabled because GITHUB_TOKEN is not set.",
"content": "GitHub integration is disabled because GITHUB_TOKEN is not set."
}
if not body or not body.strip():
raise ValueError("Comment body cannot be empty")
owner, repo = _get_github_repo_info()
headers = _get_headers()
url = f"https://api.github.com/repos/{owner}/{repo}/issues/{pr_number}/comments"
data = {"body": body}
try:
response = requests.post(url, headers=headers, json=data)
response.raise_for_status()
comment_data = response.json()
return {
"success": True,
"comment_url": comment_data["html_url"],
"content": f"Successfully added comment to PR #{pr_number}\nURL: {comment_data['html_url']}"
}
except requests.exceptions.RequestException as e:
raise RuntimeError(f"Failed to add comment to PR #{pr_number}: {str(e)}")