Add mcp-gateway/forgejo-mcp/forgejo_mcp.py

mcp-gateway/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"
+    )