From 7c8861e048484353a808b30bb521b22eea8809fe Mon Sep 17 00:00:00 2001 From: zgaetano Date: Tue, 31 Mar 2026 15:33:34 -0400 Subject: [PATCH] Add forgejo-mcp/forgejo_mcp.py --- forgejo-mcp/forgejo_mcp.py | 329 +++++++++++++++++++++++++++++++++++++ 1 file changed, 329 insertions(+) create mode 100644 forgejo-mcp/forgejo_mcp.py diff --git a/forgejo-mcp/forgejo_mcp.py b/forgejo-mcp/forgejo_mcp.py new file mode 100644 index 0000000..9a3d63a --- /dev/null +++ b/forgejo-mcp/forgejo_mcp.py @@ -0,0 +1,329 @@ +""" +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" + )