134 lines
3.0 KiB
Python
134 lines
3.0 KiB
Python
import os
|
|
import httpx
|
|
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
GITEA_URL = os.environ.get(
|
|
"GITEA_URL", "https://git.absdev.net"
|
|
).rstrip("/")
|
|
|
|
GITEA_TOKEN = os.environ.get("GITEA_TOKEN")
|
|
|
|
mcp = FastMCP(
|
|
"ABSDEV Gitea",
|
|
host="127.0.0.1",
|
|
port=8765,
|
|
)
|
|
|
|
|
|
async def api_get(path: str, params: dict | None = None):
|
|
if not GITEA_TOKEN:
|
|
raise RuntimeError("GITEA_TOKEN is not configured")
|
|
|
|
headers = {
|
|
"Authorization": f"token {GITEA_TOKEN}",
|
|
"Accept": "application/json",
|
|
}
|
|
|
|
async with httpx.AsyncClient(
|
|
base_url=GITEA_URL,
|
|
headers=headers,
|
|
timeout=30,
|
|
follow_redirects=False,
|
|
) as client:
|
|
response = await client.get(
|
|
f"/api/v1/{path.lstrip('/')}",
|
|
params=params,
|
|
)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def validate_repo(owner: str, repo: str):
|
|
import re
|
|
|
|
pattern = r"^[A-Za-z0-9_.-]+$"
|
|
|
|
if not re.fullmatch(pattern, owner):
|
|
raise ValueError("Invalid repository owner")
|
|
|
|
if not re.fullmatch(pattern, repo):
|
|
raise ValueError("Invalid repository name")
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_repositories():
|
|
"""List repositories accessible to the configured Gitea account."""
|
|
return await api_get(
|
|
"user/repos",
|
|
{"limit": 100},
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_branches(owner: str, repo: str):
|
|
"""List branches in a Gitea repository."""
|
|
validate_repo(owner, repo)
|
|
return await api_get(
|
|
f"repos/{owner}/{repo}/branches"
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
async def get_file(owner: str, repo: str, path: str, ref: str = ""):
|
|
"""Read a source file from a Gitea repository."""
|
|
from urllib.parse import quote
|
|
|
|
validate_repo(owner, repo)
|
|
|
|
if (
|
|
not path
|
|
or path.startswith("/")
|
|
or any(part in ("", ".", "..") for part in path.split("/"))
|
|
):
|
|
raise ValueError("Invalid file path")
|
|
|
|
encoded_path = quote(path, safe="/")
|
|
params = {"ref": ref} if ref else None
|
|
|
|
result = await api_get(
|
|
f"repos/{owner}/{repo}/contents/{encoded_path}",
|
|
params,
|
|
)
|
|
|
|
if isinstance(result, dict) and result.get("encoding") == "base64":
|
|
import base64
|
|
|
|
content = base64.b64decode(
|
|
result.get("content", ""),
|
|
validate=False,
|
|
)
|
|
|
|
try:
|
|
result["decoded_content"] = content.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
result["decoded_content"] = "[Binary file]"
|
|
|
|
result.pop("content", None)
|
|
|
|
return result
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_commits(owner: str, repo: str):
|
|
"""List recent commits in a repository."""
|
|
validate_repo(owner, repo)
|
|
return await api_get(
|
|
f"repos/{owner}/{repo}/commits",
|
|
{"limit": 30},
|
|
)
|
|
|
|
|
|
@mcp.tool()
|
|
async def list_pull_requests(owner: str, repo: str):
|
|
"""List pull requests in a repository."""
|
|
validate_repo(owner, repo)
|
|
return await api_get(
|
|
f"repos/{owner}/{repo}/pulls",
|
|
{"state": "all", "limit": 50},
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run(transport="streamable-http")
|