Remove mcp-gateway/forgejo-mcp/forgejo_mcp.py
This commit is contained in:
parent
aad3254d4a
commit
afcfaa86a0
1 changed files with 0 additions and 329 deletions
|
|
@ -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"
|
|
||||||
)
|
|
||||||
Loading…
Reference in a new issue