Initial commit

This commit is contained in:
2026-03-17 07:15:38 +00:00
commit ed207ab5a4
9 changed files with 2137 additions and 0 deletions

855
mcps/gitea_mcp.py Normal file
View File

@@ -0,0 +1,855 @@
"""Gitea MCP server — git-style tools for a self-hosted Gitea instance."""
from __future__ import annotations
import base64
import os
import httpx
from mcp.server.fastmcp import FastMCP
mcp = FastMCP(
"Gitea MCP",
instructions="MCP server for interacting with a Gitea instance. "
"Provides tools for repositories, issues, pull requests, files, branches, "
"releases, organizations, and users.",
)
GITEA_URL = os.environ.get("GITEA_URL", "http://localhost:3000")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
_TIMEOUT = 15.0
def _client() -> httpx.Client:
return httpx.Client(
base_url=f"{GITEA_URL}/api/v1",
headers={
"Authorization": f"token {GITEA_TOKEN}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=_TIMEOUT,
)
def _request(method: str, path: str, **kwargs) -> dict | list | str:
with _client() as c:
try:
resp = c.request(method, path, **kwargs)
resp.raise_for_status()
if resp.status_code == 204:
return {"ok": True}
return resp.json()
except httpx.HTTPStatusError as e:
body = e.response.text[:500]
return {"error": f"HTTP {e.response.status_code}", "detail": body}
except httpx.RequestError as e:
return {"error": str(e)}
def _get(path: str, **params) -> dict | list | str:
return _request("GET", path, params=params)
def _post(path: str, body: dict | None = None) -> dict | list | str:
return _request("POST", path, json=body or {})
def _patch(path: str, body: dict) -> dict | list | str:
return _request("PATCH", path, json=body)
def _delete(path: str) -> dict | list | str:
return _request("DELETE", path)
# ───────────────────────────────────────────────────────────────────
# User / Auth
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def get_authenticated_user() -> dict | list | str:
"""Get the currently authenticated Gitea user."""
return _get("/user")
# ───────────────────────────────────────────────────────────────────
# Repositories
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_repos(limit: int = 20, page: int = 1) -> dict | list | str:
"""List repositories accessible to the authenticated user.
Args:
limit: Number of results per page. Default 20.
page: Page number. Default 1.
"""
return _get("/user/repos", limit=limit, page=page)
@mcp.tool()
def get_repo(owner: str, repo: str) -> dict | list | str:
"""Get details of a repository.
Args:
owner: Repository owner username.
repo: Repository name.
"""
return _get(f"/repos/{owner}/{repo}")
@mcp.tool()
def create_repo(
name: str,
description: str = "",
private: bool = False,
auto_init: bool = True,
default_branch: str = "main",
) -> dict | list | str:
"""Create a new repository for the authenticated user.
Args:
name: Repository name.
description: Repository description.
private: Whether the repo is private. Default False.
auto_init: Initialize with a README. Default True.
default_branch: Default branch name. Default "main".
"""
return _post(
"/user/repos",
{
"name": name,
"description": description,
"private": private,
"auto_init": auto_init,
"default_branch": default_branch,
},
)
@mcp.tool()
def delete_repo(owner: str, repo: str) -> dict | list | str:
"""Delete a repository. This is irreversible.
Args:
owner: Repository owner username.
repo: Repository name.
"""
return _delete(f"/repos/{owner}/{repo}")
@mcp.tool()
def search_repos(query: str, limit: int = 10) -> dict | list | str:
"""Search for repositories.
Args:
query: Search query string.
limit: Maximum results to return. Default 10.
"""
return _get("/repos/search", q=query, limit=limit)
@mcp.tool()
def fork_repo(owner: str, repo: str, new_name: str | None = None) -> dict | list | str:
"""Fork a repository.
Args:
owner: Owner of the repository to fork.
repo: Repository name to fork.
new_name: Optional new name for the fork.
"""
body: dict = {}
if new_name:
body["name"] = new_name
return _post(f"/repos/{owner}/{repo}/forks", body)
# ───────────────────────────────────────────────────────────────────
# Branches
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_branches(owner: str, repo: str) -> dict | list | str:
"""List branches of a repository.
Args:
owner: Repository owner.
repo: Repository name.
"""
return _get(f"/repos/{owner}/{repo}/branches")
@mcp.tool()
def get_branch(owner: str, repo: str, branch: str) -> dict | list | str:
"""Get details of a specific branch.
Args:
owner: Repository owner.
repo: Repository name.
branch: Branch name.
"""
return _get(f"/repos/{owner}/{repo}/branches/{branch}")
@mcp.tool()
def create_branch(
owner: str, repo: str, branch_name: str, old_branch: str = "main"
) -> dict | list | str:
"""Create a new branch.
Args:
owner: Repository owner.
repo: Repository name.
branch_name: Name of the new branch.
old_branch: Branch to create from. Default "main".
"""
return _post(
f"/repos/{owner}/{repo}/branches",
{
"new_branch_name": branch_name,
"old_branch_name": old_branch,
},
)
@mcp.tool()
def delete_branch(owner: str, repo: str, branch: str) -> dict | list | str:
"""Delete a branch.
Args:
owner: Repository owner.
repo: Repository name.
branch: Branch name to delete.
"""
return _delete(f"/repos/{owner}/{repo}/branches/{branch}")
# ───────────────────────────────────────────────────────────────────
# File contents
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def get_file(
owner: str, repo: str, filepath: str, ref: str | None = None
) -> dict | list | str:
"""Get the contents of a file from a repository. Returns decoded text content.
Args:
owner: Repository owner.
repo: Repository name.
filepath: Path to the file in the repo.
ref: Optional branch/tag/commit to read from.
"""
params = {}
if ref:
params["ref"] = ref
result = _get(f"/repos/{owner}/{repo}/contents/{filepath}", **params)
# Decode base64 content for convenience
if (
isinstance(result, dict)
and "content" in result
and result.get("encoding") == "base64"
):
try:
result["content"] = base64.b64decode(result["content"]).decode(
"utf-8", errors="replace"
)
result["encoding"] = "utf-8"
except Exception:
pass
return result
@mcp.tool()
def create_or_update_file(
owner: str,
repo: str,
filepath: str,
content: str,
message: str,
branch: str | None = None,
sha: str | None = None,
) -> dict | list | str:
"""Create or update a file in a repository.
To update an existing file you must provide the current sha (from get_file).
Args:
owner: Repository owner.
repo: Repository name.
filepath: Path where the file will be created/updated.
content: File content (plain text, will be base64-encoded).
message: Commit message.
branch: Branch to commit to. Uses repo default if omitted.
sha: Current SHA of the file (required for updates, omit for creation).
"""
body: dict = {
"content": base64.b64encode(content.encode()).decode(),
"message": message,
}
if branch:
body["branch"] = branch
if sha:
body["sha"] = sha
method = "PUT"
with _client() as c:
try:
resp = c.request(
method, f"/repos/{owner}/{repo}/contents/{filepath}", json=body
)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
return {
"error": f"HTTP {e.response.status_code}",
"detail": e.response.text[:500],
}
except httpx.RequestError as e:
return {"error": str(e)}
@mcp.tool()
def delete_file(
owner: str,
repo: str,
filepath: str,
message: str,
sha: str,
branch: str | None = None,
) -> dict | list | str:
"""Delete a file from a repository.
Args:
owner: Repository owner.
repo: Repository name.
filepath: Path of the file to delete.
message: Commit message.
sha: Current SHA of the file (from get_file).
branch: Branch to delete from. Uses repo default if omitted.
"""
body: dict = {"message": message, "sha": sha}
if branch:
body["branch"] = branch
with _client() as c:
try:
resp = c.request(
"DELETE", f"/repos/{owner}/{repo}/contents/{filepath}", json=body
)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
return {
"error": f"HTTP {e.response.status_code}",
"detail": e.response.text[:500],
}
except httpx.RequestError as e:
return {"error": str(e)}
@mcp.tool()
def list_directory(
owner: str, repo: str, path: str = "", ref: str | None = None
) -> dict | list | str:
"""List files and directories at a path in a repository.
Args:
owner: Repository owner.
repo: Repository name.
path: Directory path. Empty string for root.
ref: Optional branch/tag/commit.
"""
params = {}
if ref:
params["ref"] = ref
return _get(f"/repos/{owner}/{repo}/contents/{path}", **params)
# ───────────────────────────────────────────────────────────────────
# Commits
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_commits(
owner: str, repo: str, sha: str | None = None, limit: int = 10, page: int = 1
) -> dict | list | str:
"""List commits in a repository.
Args:
owner: Repository owner.
repo: Repository name.
sha: Optional branch/tag/commit SHA to list from.
limit: Number of commits per page. Default 10.
page: Page number. Default 1.
"""
params: dict = {"limit": limit, "page": page}
if sha:
params["sha"] = sha
return _get(f"/repos/{owner}/{repo}/git/commits", **params)
@mcp.tool()
def get_commit(owner: str, repo: str, sha: str) -> dict | list | str:
"""Get details of a specific commit.
Args:
owner: Repository owner.
repo: Repository name.
sha: Commit SHA.
"""
return _get(f"/repos/{owner}/{repo}/git/commits/{sha}")
# ───────────────────────────────────────────────────────────────────
# Issues
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_issues(
owner: str,
repo: str,
state: str = "open",
labels: str | None = None,
limit: int = 20,
page: int = 1,
) -> dict | list | str:
"""List issues in a repository.
Args:
owner: Repository owner.
repo: Repository name.
state: Filter by state: "open", "closed", or "all". Default "open".
labels: Comma-separated label names to filter by.
limit: Results per page. Default 20.
page: Page number. Default 1.
"""
params: dict = {"state": state, "limit": limit, "page": page, "type": "issues"}
if labels:
params["labels"] = labels
return _get(f"/repos/{owner}/{repo}/issues", **params)
@mcp.tool()
def get_issue(owner: str, repo: str, index: int) -> dict | list | str:
"""Get details of an issue.
Args:
owner: Repository owner.
repo: Repository name.
index: Issue number.
"""
return _get(f"/repos/{owner}/{repo}/issues/{index}")
@mcp.tool()
def create_issue(
owner: str,
repo: str,
title: str,
body: str = "",
labels: list[int] | None = None,
assignees: list[str] | None = None,
) -> dict | list | str:
"""Create a new issue.
Args:
owner: Repository owner.
repo: Repository name.
title: Issue title.
body: Issue body/description.
labels: List of label IDs to add.
assignees: List of usernames to assign.
"""
payload: dict = {"title": title, "body": body}
if labels:
payload["labels"] = labels
if assignees:
payload["assignees"] = assignees
return _post(f"/repos/{owner}/{repo}/issues", payload)
@mcp.tool()
def edit_issue(
owner: str,
repo: str,
index: int,
title: str | None = None,
body: str | None = None,
state: str | None = None,
) -> dict | list | str:
"""Edit an existing issue.
Args:
owner: Repository owner.
repo: Repository name.
index: Issue number.
title: New title (omit to keep current).
body: New body (omit to keep current).
state: Set to "open" or "closed".
"""
payload: dict = {}
if title is not None:
payload["title"] = title
if body is not None:
payload["body"] = body
if state is not None:
payload["state"] = state
return _patch(f"/repos/{owner}/{repo}/issues/{index}", payload)
@mcp.tool()
def list_issue_comments(owner: str, repo: str, index: int) -> dict | list | str:
"""List comments on an issue.
Args:
owner: Repository owner.
repo: Repository name.
index: Issue number.
"""
return _get(f"/repos/{owner}/{repo}/issues/{index}/comments")
@mcp.tool()
def create_issue_comment(
owner: str, repo: str, index: int, body: str
) -> dict | list | str:
"""Add a comment to an issue.
Args:
owner: Repository owner.
repo: Repository name.
index: Issue number.
body: Comment text.
"""
return _post(f"/repos/{owner}/{repo}/issues/{index}/comments", {"body": body})
# ───────────────────────────────────────────────────────────────────
# Labels
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_labels(owner: str, repo: str) -> dict | list | str:
"""List labels in a repository.
Args:
owner: Repository owner.
repo: Repository name.
"""
return _get(f"/repos/{owner}/{repo}/labels")
@mcp.tool()
def create_label(
owner: str, repo: str, name: str, color: str = "#0075ca", description: str = ""
) -> dict | list | str:
"""Create a label in a repository.
Args:
owner: Repository owner.
repo: Repository name.
name: Label name.
color: Hex color code. Default "#0075ca".
description: Label description.
"""
return _post(
f"/repos/{owner}/{repo}/labels",
{
"name": name,
"color": color,
"description": description,
},
)
# ───────────────────────────────────────────────────────────────────
# Pull Requests
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_pull_requests(
owner: str, repo: str, state: str = "open", limit: int = 20, page: int = 1
) -> dict | list | str:
"""List pull requests in a repository.
Args:
owner: Repository owner.
repo: Repository name.
state: Filter by state: "open", "closed", or "all". Default "open".
limit: Results per page. Default 20.
page: Page number. Default 1.
"""
return _get(f"/repos/{owner}/{repo}/pulls", state=state, limit=limit, page=page)
@mcp.tool()
def get_pull_request(owner: str, repo: str, index: int) -> dict | list | str:
"""Get details of a pull request.
Args:
owner: Repository owner.
repo: Repository name.
index: Pull request number.
"""
return _get(f"/repos/{owner}/{repo}/pulls/{index}")
@mcp.tool()
def create_pull_request(
owner: str,
repo: str,
title: str,
head: str,
base: str = "main",
body: str = "",
) -> dict | list | str:
"""Create a new pull request.
Args:
owner: Repository owner.
repo: Repository name.
title: PR title.
head: Source branch.
base: Target branch. Default "main".
body: PR description.
"""
return _post(
f"/repos/{owner}/{repo}/pulls",
{
"title": title,
"head": head,
"base": base,
"body": body,
},
)
@mcp.tool()
def merge_pull_request(
owner: str, repo: str, index: int, merge_style: str = "merge", message: str = ""
) -> dict | list | str:
"""Merge a pull request.
Args:
owner: Repository owner.
repo: Repository name.
index: Pull request number.
merge_style: One of "merge", "rebase", "rebase-merge", "squash". Default "merge".
message: Optional merge commit message.
"""
payload: dict = {"Do": merge_style}
if message:
payload["merge_message_field"] = message
return _post(f"/repos/{owner}/{repo}/pulls/{index}/merge", payload)
@mcp.tool()
def list_pr_comments(owner: str, repo: str, index: int) -> dict | list | str:
"""List review comments on a pull request.
Args:
owner: Repository owner.
repo: Repository name.
index: Pull request number.
"""
return _get(f"/repos/{owner}/{repo}/pulls/{index}/reviews")
@mcp.tool()
def get_pr_diff(owner: str, repo: str, index: int) -> dict | list | str:
"""Get the diff of a pull request.
Args:
owner: Repository owner.
repo: Repository name.
index: Pull request number.
"""
with _client() as c:
try:
resp = c.get(f"/repos/{owner}/{repo}/pulls/{index}.diff")
resp.raise_for_status()
return {"diff": resp.text[:50000]}
except httpx.HTTPStatusError as e:
return {
"error": f"HTTP {e.response.status_code}",
"detail": e.response.text[:500],
}
except httpx.RequestError as e:
return {"error": str(e)}
# ───────────────────────────────────────────────────────────────────
# Releases / Tags
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_releases(
owner: str, repo: str, limit: int = 10, page: int = 1
) -> dict | list | str:
"""List releases in a repository.
Args:
owner: Repository owner.
repo: Repository name.
limit: Results per page. Default 10.
page: Page number. Default 1.
"""
return _get(f"/repos/{owner}/{repo}/releases", limit=limit, page=page)
@mcp.tool()
def create_release(
owner: str,
repo: str,
tag_name: str,
name: str = "",
body: str = "",
draft: bool = False,
prerelease: bool = False,
target: str | None = None,
) -> dict | list | str:
"""Create a new release.
Args:
owner: Repository owner.
repo: Repository name.
tag_name: Tag for the release (e.g. "v1.0.0").
name: Release title.
body: Release notes.
draft: Whether this is a draft release.
prerelease: Whether this is a pre-release.
target: Branch or commit SHA to tag. Uses default branch if omitted.
"""
payload: dict = {
"tag_name": tag_name,
"name": name or tag_name,
"body": body,
"draft": draft,
"prerelease": prerelease,
}
if target:
payload["target_commitish"] = target
return _post(f"/repos/{owner}/{repo}/releases", payload)
@mcp.tool()
def list_tags(owner: str, repo: str) -> dict | list | str:
"""List tags in a repository.
Args:
owner: Repository owner.
repo: Repository name.
"""
return _get(f"/repos/{owner}/{repo}/tags")
# ───────────────────────────────────────────────────────────────────
# Organizations
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_orgs() -> dict | list | str:
"""List organizations the authenticated user belongs to."""
return _get("/user/orgs")
@mcp.tool()
def get_org(org: str) -> dict | list | str:
"""Get details of an organization.
Args:
org: Organization name.
"""
return _get(f"/orgs/{org}")
@mcp.tool()
def list_org_repos(org: str, limit: int = 20, page: int = 1) -> dict | list | str:
"""List repositories in an organization.
Args:
org: Organization name.
limit: Results per page. Default 20.
page: Page number. Default 1.
"""
return _get(f"/orgs/{org}/repos", limit=limit, page=page)
# ───────────────────────────────────────────────────────────────────
# Users
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def get_user(username: str) -> dict | list | str:
"""Get a user's profile.
Args:
username: The username to look up.
"""
return _get(f"/users/{username}")
@mcp.tool()
def list_user_repos(username: str, limit: int = 20, page: int = 1) -> dict | list | str:
"""List a user's repositories.
Args:
username: The username.
limit: Results per page. Default 20.
page: Page number. Default 1.
"""
return _get(f"/users/{username}/repos", limit=limit, page=page)
# ───────────────────────────────────────────────────────────────────
# Milestones
# ───────────────────────────────────────────────────────────────────
@mcp.tool()
def list_milestones(owner: str, repo: str, state: str = "open") -> dict | list | str:
"""List milestones in a repository.
Args:
owner: Repository owner.
repo: Repository name.
state: Filter by state: "open", "closed", or "all". Default "open".
"""
return _get(f"/repos/{owner}/{repo}/milestones", state=state)
@mcp.tool()
def create_milestone(
owner: str, repo: str, title: str, description: str = ""
) -> dict | list | str:
"""Create a milestone in a repository.
Args:
owner: Repository owner.
repo: Repository name.
title: Milestone title.
description: Milestone description.
"""
return _post(
f"/repos/{owner}/{repo}/milestones",
{
"title": title,
"description": description,
},
)
if __name__ == "__main__":
mcp.run()