330 lines
9.6 KiB
Python
330 lines
9.6 KiB
Python
|
"""
|
||
|
|
Forgejo MCP Server
|
||
|
|
==================
|
||
|
|
MCP server providing tools to interact with a self-hosted Forgejo Git service.
|
||
|
|
Uses FastMCP with streamable-http transport, matching other gateway backends.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import base64
|
||
|
|
import logging
|
||
|
|
import os
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
from mcp.server.fastmcp import FastMCP
|
||
|
|
|
||
|
|
logging.basicConfig(level=logging.INFO)
|
||
|
|
logger = logging.getLogger("forgejo-mcp")
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Configuration
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
FORGEJO_URL = os.environ.get("FORGEJO_URL", "https://forge.wilddragon.net")
|
||
|
|
FORGEJO_TOKEN = os.environ.get("FORGEJO_ACCESS_TOKEN", "9b80e9ea5e1d1955585fab22fb86e740951940b4")
|
||
|
|
|
||
|
|
if not FORGEJO_TOKEN:
|
||
|
|
logger.warning("FORGEJO_ACCESS_TOKEN not set.")
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# FastMCP Server
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
mcp = FastMCP("forgejo-mcp")
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Forgejo API Client
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
def get_client() -> httpx.Client:
|
||
|
|
return httpx.Client(
|
||
|
|
base_url=FORGEJO_URL.rstrip("/"),
|
||
|
|
headers={"Authorization": f"token {FORGEJO_TOKEN}"},
|
||
|
|
timeout=30.0,
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
def api_get(endpoint: str, params: dict | None = None) -> Any:
|
||
|
|
with get_client() as client:
|
||
|
|
r = client.get(f"/api/v1{endpoint}", params=params or {})
|
||
|
|
r.raise_for_status()
|
||
|
|
return r.json() if r.status_code != 204 else {}
|
||
|
|
|
||
|
|
|
||
|
|
def api_post(endpoint: str, body: dict | None = None) -> Any:
|
||
|
|
with get_client() as client:
|
||
|
|
r = client.post(f"/api/v1{endpoint}", json=body or {})
|
||
|
|
r.raise_for_status()
|
||
|
|
return r.json() if r.status_code != 204 else {}
|
||
|
|
|
||
|
|
|
||
|
|
def api_patch(endpoint: str, body: dict | None = None) -> Any:
|
||
|
|
with get_client() as client:
|
||
|
|
r = client.patch(f"/api/v1{endpoint}", json=body or {})
|
||
|
|
r.raise_for_status()
|
||
|
|
return r.json() if r.status_code != 204 else {}
|
||
|
|
|
||
|
|
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
# Tools
|
||
|
|
# ---------------------------------------------------------------------------
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_list_repositories(owner: str, limit: int = 20, page: int = 1) -> str:
|
||
|
|
"""List repositories for a user or organization.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Username or organization name
|
||
|
|
limit: Max repositories to return (default 20)
|
||
|
|
page: Page number for pagination (default 1)
|
||
|
|
"""
|
||
|
|
repos = api_get(f"/users/{owner}/repos", {"limit": limit, "page": page})
|
||
|
|
if not repos:
|
||
|
|
return "No repositories found."
|
||
|
|
lines = [f"- {r['name']}: {r.get('description', 'No description')}" for r in repos]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_get_repository(owner: str, repo: str) -> str:
|
||
|
|
"""Get details about a specific repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner (username or organization)
|
||
|
|
repo: Repository name
|
||
|
|
"""
|
||
|
|
r = api_get(f"/repos/{owner}/{repo}")
|
||
|
|
return (
|
||
|
|
f"Repository: {r['name']}\n"
|
||
|
|
f"Owner: {r['owner']['login']}\n"
|
||
|
|
f"Description: {r.get('description', 'None')}\n"
|
||
|
|
f"URL: {r['html_url']}\n"
|
||
|
|
f"Stars: {r['stargazers_count']}\n"
|
||
|
|
f"Forks: {r['forks_count']}\n"
|
||
|
|
f"Private: {r['private']}\n"
|
||
|
|
f"Default Branch: {r['default_branch']}\n"
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_create_repository(
|
||
|
|
name: str,
|
||
|
|
description: str = "",
|
||
|
|
private: bool = False,
|
||
|
|
auto_init: bool = True,
|
||
|
|
) -> str:
|
||
|
|
"""Create a new repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
name: Repository name (lowercase, no spaces)
|
||
|
|
description: Repository description
|
||
|
|
private: Whether the repository is private (default false)
|
||
|
|
auto_init: Initialize with a README (default true)
|
||
|
|
"""
|
||
|
|
repo = api_post("/user/repos", {
|
||
|
|
"name": name,
|
||
|
|
"description": description,
|
||
|
|
"private": private,
|
||
|
|
"auto_init": auto_init,
|
||
|
|
})
|
||
|
|
return f"Repository created: {repo['html_url']}"
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_list_issues(
|
||
|
|
owner: str,
|
||
|
|
repo: str,
|
||
|
|
state: str = "open",
|
||
|
|
limit: int = 20,
|
||
|
|
page: int = 1,
|
||
|
|
) -> str:
|
||
|
|
"""List issues in a repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
state: Filter by state: open, closed, all (default open)
|
||
|
|
limit: Max issues to return (default 20)
|
||
|
|
page: Page number (default 1)
|
||
|
|
"""
|
||
|
|
issues = api_get(f"/repos/{owner}/{repo}/issues", {"state": state, "limit": limit, "page": page})
|
||
|
|
if not issues:
|
||
|
|
return "No issues found."
|
||
|
|
lines = [f"#{i['number']}: {i['title']} ({i['state']})" for i in issues]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_create_issue(
|
||
|
|
owner: str,
|
||
|
|
repo: str,
|
||
|
|
title: str,
|
||
|
|
body: str = "",
|
||
|
|
labels: list[str] | None = None,
|
||
|
|
) -> str:
|
||
|
|
"""Create a new issue in a repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
title: Issue title
|
||
|
|
body: Issue description (markdown)
|
||
|
|
labels: List of label names
|
||
|
|
"""
|
||
|
|
issue = api_post(f"/repos/{owner}/{repo}/issues", {
|
||
|
|
"title": title,
|
||
|
|
"body": body,
|
||
|
|
"labels": labels or [],
|
||
|
|
})
|
||
|
|
return f"Issue created: #{issue['number']} - {issue['title']}\nURL: {issue['html_url']}"
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_update_issue(
|
||
|
|
owner: str,
|
||
|
|
repo: str,
|
||
|
|
issue_number: int,
|
||
|
|
state: str | None = None,
|
||
|
|
title: str | None = None,
|
||
|
|
body: str | None = None,
|
||
|
|
) -> str:
|
||
|
|
"""Update an existing issue.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
issue_number: Issue number to update
|
||
|
|
state: New state: open or closed
|
||
|
|
title: New issue title
|
||
|
|
body: New issue description
|
||
|
|
"""
|
||
|
|
payload = {k: v for k, v in {"state": state, "title": title, "body": body}.items() if v is not None}
|
||
|
|
issue = api_patch(f"/repos/{owner}/{repo}/issues/{issue_number}", payload)
|
||
|
|
return f"Issue updated: #{issue['number']} - {issue['title']}"
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_list_pull_requests(
|
||
|
|
owner: str,
|
||
|
|
repo: str,
|
||
|
|
state: str = "open",
|
||
|
|
limit: int = 20,
|
||
|
|
page: int = 1,
|
||
|
|
) -> str:
|
||
|
|
"""List pull requests in a repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
state: Filter by state: open, closed, all (default open)
|
||
|
|
limit: Max PRs to return (default 20)
|
||
|
|
page: Page number (default 1)
|
||
|
|
"""
|
||
|
|
prs = api_get(f"/repos/{owner}/{repo}/pulls", {"state": state, "limit": limit, "page": page})
|
||
|
|
if not prs:
|
||
|
|
return "No pull requests found."
|
||
|
|
lines = [f"#{p['number']}: {p['title']} ({p['state']})" for p in prs]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_create_pull_request(
|
||
|
|
owner: str,
|
||
|
|
repo: str,
|
||
|
|
title: str,
|
||
|
|
head: str,
|
||
|
|
base: str,
|
||
|
|
body: str = "",
|
||
|
|
) -> str:
|
||
|
|
"""Create a new pull request.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
title: PR title
|
||
|
|
head: Head branch (source branch)
|
||
|
|
base: Base branch (target branch)
|
||
|
|
body: PR description
|
||
|
|
"""
|
||
|
|
pr = api_post(f"/repos/{owner}/{repo}/pulls", {
|
||
|
|
"title": title,
|
||
|
|
"head": head,
|
||
|
|
"base": base,
|
||
|
|
"body": body,
|
||
|
|
})
|
||
|
|
return f"Pull request created: #{pr['number']}\nURL: {pr['html_url']}"
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_list_branches(owner: str, repo: str, limit: int = 20, page: int = 1) -> str:
|
||
|
|
"""List branches in a repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
limit: Max branches to return (default 20)
|
||
|
|
page: Page number (default 1)
|
||
|
|
"""
|
||
|
|
branches = api_get(f"/repos/{owner}/{repo}/branches", {"limit": limit, "page": page})
|
||
|
|
if not branches:
|
||
|
|
return "No branches found."
|
||
|
|
lines = [f"- {b['name']} (commit: {b['commit']['sha'][:7]})" for b in branches]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_get_file(owner: str, repo: str, path: str, ref: str = "main") -> str:
|
||
|
|
"""Get file contents from a repository.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
owner: Repository owner
|
||
|
|
repo: Repository name
|
||
|
|
path: File path within the repository
|
||
|
|
ref: Branch, tag, or commit SHA (default main)
|
||
|
|
"""
|
||
|
|
content = api_get(f"/repos/{owner}/{repo}/contents/{path}", {"ref": ref})
|
||
|
|
if isinstance(content.get("content"), str):
|
||
|
|
file_content = base64.b64decode(content["content"]).decode("utf-8")
|
||
|
|
else:
|
||
|
|
file_content = content.get("content", "")
|
||
|
|
return f"File: {path}\n\n{file_content}"
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_search_repositories(q: str, limit: int = 20, page: int = 1) -> str:
|
||
|
|
"""Search for repositories across the Forgejo instance.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
q: Search query
|
||
|
|
limit: Max results to return (default 20)
|
||
|
|
page: Page number (default 1)
|
||
|
|
"""
|
||
|
|
resp = api_get("/repos/search", {"q": q, "limit": limit, "page": page})
|
||
|
|
repos = resp.get("data", [])
|
||
|
|
if not repos:
|
||
|
|
return "No repositories found."
|
||
|
|
lines = [f"- {r['full_name']}: {r.get('description', 'No description')}" for r in repos]
|
||
|
|
return "\n".join(lines)
|
||
|
|
|
||
|
|
|
||
|
|
@mcp.tool()
|
||
|
|
def forgejo_get_user(username: str) -> str:
|
||
|
|
"""Get user profile information.
|
||
|
|
|
||
|
|
Args:
|
||
|
|
username: The Forgejo username to look up
|
||
|
|
"""
|
||
|
|
u = api_get(f"/users/{username}")
|
||
|
|
return (
|
||
|
|
f"User: {u['login']}\n"
|
||
|
|
f"Full Name: {u.get('full_name', 'N/A')}\n"
|
||
|
|
f"Email: {u.get('email', 'N/A')}\n"
|
||
|
|
f"Bio: {u.get('description', 'N/A')}\n"
|
||
|
|
f"Repositories: {u.get('repo_count', 0)}\n"
|
||
|
|
f"Followers: {u.get('followers_count', 0)}\n"
|
||
|
|
f"Following: {u.get('following_count', 0)}\n"
|
||
|
|
f"Profile URL: {u['html_url']}\n"
|
||
|
|
)
|