a-limasov-ii
781439bbdb
- Full Gitea REST API coverage for repos, issues, PRs, branches, files, releases, webhooks, orgs, notifications - Multi-instance support via config.json - pyproject.toml with pinned dependencies (mcp[cli], httpx) - config.example.json as credentials template (config.json is gitignored)
648 lines
28 KiB
Python
648 lines
28 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Gitea MCP Server — full Gitea platform management via REST API.
|
|
Reads credentials from config.json (passed via --config argument or auto-discovered).
|
|
"""
|
|
|
|
import base64
|
|
import os
|
|
import sys
|
|
import json
|
|
import argparse
|
|
import httpx
|
|
from typing import Any
|
|
from urllib.parse import quote
|
|
from mcp.server.fastmcp import FastMCP
|
|
|
|
# ── Config loading ─────────────────────────────────────────────────────────────
|
|
|
|
def find_config() -> str:
|
|
"""Find config.json: --config arg > GITEA_CONFIG env > script dir > cwd."""
|
|
parser = argparse.ArgumentParser(add_help=False)
|
|
parser.add_argument("--config", default="")
|
|
args, _ = parser.parse_known_args()
|
|
if args.config:
|
|
return args.config
|
|
if os.environ.get("GITEA_CONFIG"):
|
|
return os.environ["GITEA_CONFIG"]
|
|
# Look next to the script
|
|
script_dir = os.path.dirname(os.path.abspath(__file__))
|
|
candidate = os.path.join(script_dir, "..", "config.json")
|
|
if os.path.exists(candidate):
|
|
return os.path.normpath(candidate)
|
|
return "config.json"
|
|
|
|
CONFIG_PATH = find_config()
|
|
|
|
try:
|
|
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
|
_config = json.load(f)
|
|
except FileNotFoundError:
|
|
print(f"ERROR: config.json not found at {CONFIG_PATH}", file=sys.stderr)
|
|
print("Create config.json next to the plugin. See README.md for format.", file=sys.stderr)
|
|
sys.exit(1)
|
|
except json.JSONDecodeError as e:
|
|
print(f"ERROR: config.json is invalid JSON: {e}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
INSTANCES: dict = _config.get("instances", {})
|
|
DEFAULT_INSTANCE: str = _config.get("default", "default")
|
|
|
|
if not INSTANCES:
|
|
print("ERROR: config.json has no 'instances' defined.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
# ── HTTP helpers ───────────────────────────────────────────────────────────────
|
|
|
|
_http = httpx.Client(timeout=30)
|
|
|
|
def _b(s: str) -> str:
|
|
"""URL-encode a branch/tag/path segment so slashes don't break the URL."""
|
|
return quote(s, safe="")
|
|
|
|
def _get_instance(instance: str) -> tuple[str, dict]:
|
|
name = instance or DEFAULT_INSTANCE
|
|
cfg = INSTANCES.get(name)
|
|
if not cfg:
|
|
available = ", ".join(INSTANCES.keys())
|
|
raise ValueError(f"Instance '{name}' not found in config.json. Available: {available}")
|
|
url = cfg["url"].rstrip("/")
|
|
headers = {
|
|
"Authorization": f"token {cfg['token']}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
}
|
|
return f"{url}/api/v1", headers
|
|
|
|
def _req(method: str, api_base: str, headers: dict, path: str, **kwargs) -> Any:
|
|
url = f"{api_base}{path}"
|
|
r = _http.request(method, url, headers=headers, **kwargs)
|
|
if r.status_code == 204:
|
|
return {"success": True}
|
|
if not r.is_success:
|
|
try:
|
|
detail = r.json()
|
|
except Exception:
|
|
detail = r.text
|
|
raise RuntimeError(f"Gitea API {r.status_code}: {detail}")
|
|
if not r.text:
|
|
return {"success": True}
|
|
return r.json()
|
|
|
|
def _call(method: str, instance: str, path: str, **kwargs) -> Any:
|
|
api, headers = _get_instance(instance)
|
|
return _req(method, api, headers, path, **kwargs)
|
|
|
|
def _strip(d: dict) -> dict:
|
|
return {k: v for k, v in d.items() if v is not None}
|
|
|
|
def GET(instance, path, params=None): return _call("GET", instance, path, params=params)
|
|
def POST(instance, path, body=None): return _call("POST", instance, path, json=body)
|
|
def PATCH(instance, path, body=None): return _call("PATCH", instance, path, json=body)
|
|
def PUT(instance, path, body=None): return _call("PUT", instance, path, json=body)
|
|
def DELETE(instance, path, body=None):
|
|
return _call("DELETE", instance, path, **({} if body is None else {"json": body}))
|
|
|
|
# ── MCP server ─────────────────────────────────────────────────────────────────
|
|
|
|
mcp = FastMCP("gitea")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# META
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_instances() -> Any:
|
|
"""List all configured Gitea instances and show which one is default."""
|
|
return {
|
|
"instances": list(INSTANCES.keys()),
|
|
"default": DEFAULT_INSTANCE,
|
|
}
|
|
|
|
@mcp.tool()
|
|
def get_server_info(instance: str = "") -> Any:
|
|
"""Get Gitea server version and info."""
|
|
return GET(instance, "/version")
|
|
|
|
@mcp.tool()
|
|
def get_current_user(instance: str = "") -> Any:
|
|
"""Get information about the authenticated user."""
|
|
return GET(instance, "/user")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# REPOSITORIES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_my_repos(limit: int = 50, page: int = 1, instance: str = "") -> Any:
|
|
"""List repositories of the authenticated user."""
|
|
return GET(instance, "/user/repos", {"limit": limit, "page": page})
|
|
|
|
@mcp.tool()
|
|
def search_repos(query: str, limit: int = 20, page: int = 1, instance: str = "") -> Any:
|
|
"""Search repositories across the Gitea instance."""
|
|
return GET(instance, "/repos/search", {"q": query, "limit": limit, "page": page})
|
|
|
|
@mcp.tool()
|
|
def get_repo(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""Get details of a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}")
|
|
|
|
@mcp.tool()
|
|
def create_repo(
|
|
name: str,
|
|
description: str = "",
|
|
private: bool = True,
|
|
auto_init: bool = True,
|
|
default_branch: str = "main",
|
|
gitignores: str = "",
|
|
license: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a new repository for the authenticated user."""
|
|
return POST(instance, "/user/repos", _strip({
|
|
"name": name,
|
|
"description": description or None,
|
|
"private": private,
|
|
"auto_init": auto_init,
|
|
"default_branch": default_branch,
|
|
"gitignores": gitignores or None,
|
|
"license": license or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def create_org_repo(
|
|
org: str,
|
|
name: str,
|
|
description: str = "",
|
|
private: bool = True,
|
|
auto_init: bool = True,
|
|
default_branch: str = "main",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a new repository in an organization."""
|
|
return POST(instance, f"/orgs/{org}/repos", _strip({
|
|
"name": name,
|
|
"description": description or None,
|
|
"private": private,
|
|
"auto_init": auto_init,
|
|
"default_branch": default_branch,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def update_repo(
|
|
owner: str, repo: str,
|
|
description: str = "", private: bool | None = None, website: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Update repository settings."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}", _strip({
|
|
"description": description or None,
|
|
"private": private,
|
|
"website": website or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def delete_repo(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""Delete a repository permanently."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}")
|
|
|
|
@mcp.tool()
|
|
def fork_repo(owner: str, repo: str, organization: str = "", new_repo_name: str = "", instance: str = "") -> Any:
|
|
"""Fork a repository."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/forks", _strip({
|
|
"organization": organization or None,
|
|
"name": new_repo_name or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def set_repo_topics(owner: str, repo: str, topics: list[str], instance: str = "") -> Any:
|
|
"""Replace all topics of a repository."""
|
|
return PUT(instance, f"/repos/{owner}/{repo}/topics", {"topics": topics})
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# BRANCHES & TAGS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_branches(owner: str, repo: str, limit: int = 50, instance: str = "") -> Any:
|
|
"""List all branches in a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/branches", {"limit": limit})
|
|
|
|
@mcp.tool()
|
|
def get_branch(owner: str, repo: str, branch: str, instance: str = "") -> Any:
|
|
"""Get details of a branch including its latest commit."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/branches/{_b(branch)}")
|
|
|
|
@mcp.tool()
|
|
def create_branch(owner: str, repo: str, new_branch_name: str, old_branch_name: str = "", instance: str = "") -> Any:
|
|
"""Create a branch. old_branch_name defaults to the repo's default branch."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/branches", _strip({
|
|
"new_branch_name": new_branch_name,
|
|
"old_branch_name": old_branch_name or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def delete_branch(owner: str, repo: str, branch: str, instance: str = "") -> Any:
|
|
"""Delete a branch."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}/branches/{_b(branch)}")
|
|
|
|
@mcp.tool()
|
|
def list_tags(owner: str, repo: str, limit: int = 20, instance: str = "") -> Any:
|
|
"""List tags in a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/tags", {"limit": limit})
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# FILES & CONTENTS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_directory(owner: str, repo: str, path: str = "", ref: str = "", instance: str = "") -> Any:
|
|
"""List files and directories at a path. ref can be branch/tag/commit."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/contents/{path}", _strip({"ref": ref or None}))
|
|
|
|
@mcp.tool()
|
|
def get_file(owner: str, repo: str, filepath: str, ref: str = "", instance: str = "") -> Any:
|
|
"""Get file content and metadata. Returns base64-encoded content + SHA."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/contents/{filepath}", _strip({"ref": ref or None}))
|
|
|
|
@mcp.tool()
|
|
def create_file(
|
|
owner: str, repo: str, filepath: str,
|
|
content: str, message: str,
|
|
branch: str = "", instance: str = "",
|
|
) -> Any:
|
|
"""Create a new text file. content is plain text — encoding handled automatically."""
|
|
encoded = base64.b64encode(content.encode()).decode()
|
|
return POST(instance, f"/repos/{owner}/{repo}/contents/{filepath}", _strip({
|
|
"content": encoded,
|
|
"message": message,
|
|
"branch": branch or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def update_file(
|
|
owner: str, repo: str, filepath: str,
|
|
content: str, message: str, sha: str,
|
|
branch: str = "", instance: str = "",
|
|
) -> Any:
|
|
"""Update an existing file. sha is the current blob SHA (from get_file). content is plain text."""
|
|
encoded = base64.b64encode(content.encode()).decode()
|
|
return PUT(instance, f"/repos/{owner}/{repo}/contents/{filepath}", _strip({
|
|
"content": encoded,
|
|
"message": message,
|
|
"sha": sha,
|
|
"branch": branch or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def delete_file(
|
|
owner: str, repo: str, filepath: str,
|
|
message: str, sha: str,
|
|
branch: str = "", instance: str = "",
|
|
) -> Any:
|
|
"""Delete a file. sha is the current blob SHA (from get_file)."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}/contents/{filepath}", _strip({
|
|
"message": message, "sha": sha, "branch": branch or None,
|
|
}))
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# ISSUES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_issues(
|
|
owner: str, repo: str,
|
|
state: str = "open", issue_type: str = "issues",
|
|
labels: str = "", assignee: str = "",
|
|
limit: int = 20, page: int = 1,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""List issues or PRs. state: open|closed|all. issue_type: issues|pulls."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/issues", _strip({
|
|
"state": state, "type": issue_type,
|
|
"labels": labels or None, "assignee": assignee or None,
|
|
"limit": limit, "page": page,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def get_issue(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""Get a single issue by number."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/issues/{index}")
|
|
|
|
@mcp.tool()
|
|
def create_issue(
|
|
owner: str, repo: str, title: str, body: str = "",
|
|
assignees: list[str] | None = None, labels: list[int] | None = None, milestone: int = 0,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a new issue."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/issues", _strip({
|
|
"title": title,
|
|
"body": body or None,
|
|
"assignees": assignees or None,
|
|
"labels": labels or None,
|
|
"milestone": milestone or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def update_issue(
|
|
owner: str, repo: str, index: int,
|
|
title: str = "", body: str = "", state: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Update an issue. state: open|closed."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}/issues/{index}", _strip({
|
|
"title": title or None, "body": body or None, "state": state or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def close_issue(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""Close an issue."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}/issues/{index}", {"state": "closed"})
|
|
|
|
@mcp.tool()
|
|
def reopen_issue(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""Reopen a closed issue."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}/issues/{index}", {"state": "open"})
|
|
|
|
@mcp.tool()
|
|
def list_issue_comments(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""List comments on an issue."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/issues/{index}/comments")
|
|
|
|
@mcp.tool()
|
|
def add_issue_comment(owner: str, repo: str, index: int, body: str, instance: str = "") -> Any:
|
|
"""Add a comment to an issue or PR."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/issues/{index}/comments", {"body": body})
|
|
|
|
@mcp.tool()
|
|
def edit_issue_comment(owner: str, repo: str, comment_id: int, body: str, instance: str = "") -> Any:
|
|
"""Edit an existing comment."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}/issues/comments/{comment_id}", {"body": body})
|
|
|
|
@mcp.tool()
|
|
def delete_issue_comment(owner: str, repo: str, comment_id: int, instance: str = "") -> Any:
|
|
"""Delete a comment."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}/issues/comments/{comment_id}")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# LABELS & MILESTONES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_labels(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""List all labels in a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/labels")
|
|
|
|
@mcp.tool()
|
|
def create_label(owner: str, repo: str, name: str, color: str, description: str = "", instance: str = "") -> Any:
|
|
"""Create a label. color is hex like #ff0000."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/labels", _strip({
|
|
"name": name, "color": color, "description": description or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def add_issue_labels(owner: str, repo: str, index: int, label_ids: list[int], instance: str = "") -> Any:
|
|
"""Add labels to an issue."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/issues/{index}/labels", {"labels": label_ids})
|
|
|
|
@mcp.tool()
|
|
def list_milestones(owner: str, repo: str, state: str = "open", instance: str = "") -> Any:
|
|
"""List milestones. state: open|closed|all."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/milestones", {"state": state})
|
|
|
|
@mcp.tool()
|
|
def create_milestone(owner: str, repo: str, title: str, description: str = "", due_on: str = "", instance: str = "") -> Any:
|
|
"""Create a milestone. due_on: 2024-12-31T00:00:00Z"""
|
|
return POST(instance, f"/repos/{owner}/{repo}/milestones", _strip({
|
|
"title": title, "description": description or None, "due_on": due_on or None,
|
|
}))
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# PULL REQUESTS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_pull_requests(owner: str, repo: str, state: str = "open", limit: int = 20, instance: str = "") -> Any:
|
|
"""List pull requests. state: open|closed|all."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/pulls", {"state": state, "limit": limit})
|
|
|
|
@mcp.tool()
|
|
def get_pull_request(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""Get a pull request by index."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/pulls/{index}")
|
|
|
|
@mcp.tool()
|
|
def create_pull_request(
|
|
owner: str, repo: str, title: str, head: str, base: str,
|
|
body: str = "", assignees: list[str] | None = None, labels: list[int] | None = None,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a pull request. head = source branch, base = target branch."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/pulls", _strip({
|
|
"title": title, "head": head, "base": base,
|
|
"body": body or None,
|
|
"assignees": assignees or None,
|
|
"labels": labels or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def update_pull_request(
|
|
owner: str, repo: str, index: int,
|
|
title: str = "", body: str = "", state: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Update a pull request."""
|
|
return PATCH(instance, f"/repos/{owner}/{repo}/pulls/{index}", _strip({
|
|
"title": title or None, "body": body or None, "state": state or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def merge_pull_request(
|
|
owner: str, repo: str, index: int,
|
|
merge_style: str = "merge",
|
|
commit_title: str = "", commit_message: str = "",
|
|
delete_branch_after_merge: bool = False,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Merge a pull request. merge_style: merge|rebase|squash|fast-forward-only."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/pulls/{index}/merge", _strip({
|
|
"Do": merge_style,
|
|
"MergeTitleField": commit_title or None,
|
|
"MergeMessageField": commit_message or None,
|
|
"delete_branch_after_merge": delete_branch_after_merge,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def get_pr_diff(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""Get the diff of a pull request as text."""
|
|
api, headers = _get_instance(instance)
|
|
url = f"{api}/repos/{owner}/{repo}/pulls/{index}.diff"
|
|
r = _http.get(url, headers=headers)
|
|
if not r.is_success:
|
|
raise RuntimeError(f"Gitea API {r.status_code}: {r.text}")
|
|
return {"diff": r.text}
|
|
|
|
@mcp.tool()
|
|
def list_pr_reviews(owner: str, repo: str, index: int, instance: str = "") -> Any:
|
|
"""List reviews on a pull request."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/pulls/{index}/reviews")
|
|
|
|
@mcp.tool()
|
|
def create_pr_review(
|
|
owner: str, repo: str, index: int,
|
|
state: str, body: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Submit a PR review. state: APPROVED|COMMENT|REQUEST_CHANGES."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/pulls/{index}/reviews", _strip({
|
|
"event": state, "body": body or None,
|
|
}))
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# RELEASES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_releases(owner: str, repo: str, limit: int = 10, instance: str = "") -> Any:
|
|
"""List releases."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/releases", {"limit": limit})
|
|
|
|
@mcp.tool()
|
|
def get_latest_release(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""Get the latest release."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/releases/latest")
|
|
|
|
@mcp.tool()
|
|
def create_release(
|
|
owner: str, repo: str, tag_name: str, name: str,
|
|
body: str = "", draft: bool = False, prerelease: bool = False,
|
|
target_commitish: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a release."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/releases", _strip({
|
|
"tag_name": tag_name, "name": name,
|
|
"body": body or None, "draft": draft, "prerelease": prerelease,
|
|
"target_commitish": target_commitish or None,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def delete_release(owner: str, repo: str, release_id: int, instance: str = "") -> Any:
|
|
"""Delete a release."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}/releases/{release_id}")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# WEBHOOKS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_repo_webhooks(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""List webhooks for a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/hooks")
|
|
|
|
@mcp.tool()
|
|
def create_repo_webhook(
|
|
owner: str, repo: str, url: str, events: list[str],
|
|
secret: str = "", active: bool = True,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Create a webhook. events: push, issues, pull_request, release, etc."""
|
|
return POST(instance, f"/repos/{owner}/{repo}/hooks", {
|
|
"type": "gitea",
|
|
"config": _strip({"url": url, "content_type": "json", "secret": secret or None}),
|
|
"events": events,
|
|
"active": active,
|
|
})
|
|
|
|
@mcp.tool()
|
|
def delete_repo_webhook(owner: str, repo: str, hook_id: int, instance: str = "") -> Any:
|
|
"""Delete a webhook."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}/hooks/{hook_id}")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# USERS & ORGANIZATIONS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def get_user(username: str, instance: str = "") -> Any:
|
|
"""Get public information about a user."""
|
|
return GET(instance, f"/users/{username}")
|
|
|
|
@mcp.tool()
|
|
def search_users(query: str, limit: int = 10, instance: str = "") -> Any:
|
|
"""Search for users."""
|
|
return GET(instance, "/users/search", {"q": query, "limit": limit})
|
|
|
|
@mcp.tool()
|
|
def list_my_orgs(instance: str = "") -> Any:
|
|
"""List organizations the authenticated user belongs to."""
|
|
return GET(instance, "/user/orgs")
|
|
|
|
@mcp.tool()
|
|
def get_org(org: str, instance: str = "") -> Any:
|
|
"""Get information about an organization."""
|
|
return GET(instance, f"/orgs/{org}")
|
|
|
|
@mcp.tool()
|
|
def list_org_repos(org: str, limit: int = 50, instance: str = "") -> Any:
|
|
"""List repositories of an organization."""
|
|
return GET(instance, f"/orgs/{org}/repos", {"limit": limit})
|
|
|
|
@mcp.tool()
|
|
def list_org_members(org: str, instance: str = "") -> Any:
|
|
"""List members of an organization."""
|
|
return GET(instance, f"/orgs/{org}/members")
|
|
|
|
@mcp.tool()
|
|
def list_org_teams(org: str, instance: str = "") -> Any:
|
|
"""List teams in an organization."""
|
|
return GET(instance, f"/orgs/{org}/teams")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# COMMITS & HISTORY
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_commits(
|
|
owner: str, repo: str,
|
|
sha: str = "", path: str = "",
|
|
limit: int = 20, page: int = 1,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""List commits. sha can be branch/tag/commit hash."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/commits", _strip({
|
|
"sha": sha or None, "path": path or None,
|
|
"limit": limit, "page": page,
|
|
}))
|
|
|
|
@mcp.tool()
|
|
def compare_branches(owner: str, repo: str, base: str, head: str, instance: str = "") -> Any:
|
|
"""Compare two branches, tags, or commits."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/compare/{_b(base)}...{_b(head)}")
|
|
|
|
@mcp.tool()
|
|
def list_repo_contributors(owner: str, repo: str, limit: int = 20, instance: str = "") -> Any:
|
|
"""List contributors to a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/contributors", {"limit": limit})
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# NOTIFICATIONS
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@mcp.tool()
|
|
def list_notifications(include_read: bool = False, limit: int = 20, instance: str = "") -> Any:
|
|
"""List notifications. include_read=True includes already-read notifications."""
|
|
return GET(instance, "/notifications", _strip({"all": include_read, "limit": limit}))
|
|
|
|
@mcp.tool()
|
|
def mark_all_notifications_read(instance: str = "") -> Any:
|
|
"""Mark all notifications as read."""
|
|
return PUT(instance, "/notifications", {})
|
|
|
|
# ──────────────────────────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
mcp.run()
|