diff --git a/mcp-gateway/forgejo-mcp/forgejo_mcp.py b/mcp-gateway/forgejo-mcp/forgejo_mcp.py deleted file mode 100644 index 9a3d63a..0000000 --- a/mcp-gateway/forgejo-mcp/forgejo_mcp.py +++ /dev/null @@ -1,329 +0,0 @@ -""" -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" - )