a-limasov-ii
9a5750ddca
Expose the same 66 Gitea tools two ways from server/gitea_mcp.py:
- MCP server (no subcommand -> mcp.run()), as before
- CLI: `<tool> --flags` prints one JSON object {success, data/error}
A new @tool decorator registers each function in both FastMCP and a TOOLS
registry; a CLI dispatcher builds argparse from each function's signature and
wraps results in the repo's JSON+success contract. Adds `list-tools` for
discovery. No tool logic duplicated.
- Rewrite skills/gitea/SKILL.md for CLI-first usage (MCP kept as an option)
- Sync version to 0.3.0 across plugin.json, pyproject.toml, SKILL.md
- Add tests/unit/test_cli.py (dispatch routing, flag typing, errors, main)
- Document dual-mode in README.md and CLAUDE.md
- Ignore coverage artifacts
Tests: 119 passed, 93% coverage.
837 lines
34 KiB
Python
837 lines
34 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 types
|
|
import inspect
|
|
import typing
|
|
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")
|
|
|
|
# Single registry of every tool, keyed by name. Populated by the @tool decorator
|
|
# below and consumed by the CLI dispatcher at the bottom of the file. This is what
|
|
# lets one source file serve both as an MCP server and as a CLI-backed skill.
|
|
TOOLS: dict[str, Any] = {}
|
|
|
|
|
|
def tool(fn):
|
|
"""Register a function as BOTH an MCP tool and a CLI subcommand.
|
|
|
|
The wrapped function stays directly callable (mcp.tool() returns the original
|
|
function), so unit tests and the CLI dispatcher can invoke it without going
|
|
through the MCP transport.
|
|
"""
|
|
TOOLS[fn.__name__] = fn
|
|
return mcp.tool()(fn)
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# META
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@tool
|
|
def list_instances() -> Any:
|
|
"""List all configured Gitea instances and show which one is default."""
|
|
return {
|
|
"instances": list(INSTANCES.keys()),
|
|
"default": DEFAULT_INSTANCE,
|
|
}
|
|
|
|
@tool
|
|
def get_server_info(instance: str = "") -> Any:
|
|
"""Get Gitea server version and info."""
|
|
return GET(instance, "/version")
|
|
|
|
@tool
|
|
def get_current_user(instance: str = "") -> Any:
|
|
"""Get information about the authenticated user."""
|
|
return GET(instance, "/user")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# REPOSITORIES
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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})
|
|
|
|
@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})
|
|
|
|
@tool
|
|
def get_repo(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""Get details of a repository."""
|
|
return GET(instance, f"/repos/{owner}/{repo}")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@tool
|
|
def delete_repo(owner: str, repo: str, instance: str = "") -> Any:
|
|
"""Delete a repository permanently."""
|
|
return DELETE(instance, f"/repos/{owner}/{repo}")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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})
|
|
|
|
@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)}")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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)}")
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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}))
|
|
|
|
@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}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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}")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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"})
|
|
|
|
@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"})
|
|
|
|
@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")
|
|
|
|
@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})
|
|
|
|
@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})
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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})
|
|
|
|
@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})
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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})
|
|
|
|
@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}")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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}
|
|
|
|
@tool
|
|
def list_pr_files(
|
|
owner: str, repo: str, index: int,
|
|
limit: int = 50, page: int = 1,
|
|
whitespace: str = "",
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""List files changed in a pull request. whitespace: ignore-all|ignore-change|ignore-eol|show-all."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/pulls/{index}/files", _strip({
|
|
"limit": limit, "page": page,
|
|
"whitespace": whitespace or None,
|
|
}))
|
|
|
|
@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")
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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})
|
|
|
|
@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")
|
|
|
|
@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,
|
|
}))
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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")
|
|
|
|
@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,
|
|
})
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@tool
|
|
def get_user(username: str, instance: str = "") -> Any:
|
|
"""Get public information about a user."""
|
|
return GET(instance, f"/users/{username}")
|
|
|
|
@tool
|
|
def search_users(query: str, limit: int = 10, instance: str = "") -> Any:
|
|
"""Search for users."""
|
|
return GET(instance, "/users/search", {"q": query, "limit": limit})
|
|
|
|
@tool
|
|
def list_my_orgs(instance: str = "") -> Any:
|
|
"""List organizations the authenticated user belongs to."""
|
|
return GET(instance, "/user/orgs")
|
|
|
|
@tool
|
|
def get_org(org: str, instance: str = "") -> Any:
|
|
"""Get information about an organization."""
|
|
return GET(instance, f"/orgs/{org}")
|
|
|
|
@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})
|
|
|
|
@tool
|
|
def list_org_members(org: str, instance: str = "") -> Any:
|
|
"""List members of an organization."""
|
|
return GET(instance, f"/orgs/{org}/members")
|
|
|
|
@tool
|
|
def list_org_teams(org: str, instance: str = "") -> Any:
|
|
"""List teams in an organization."""
|
|
return GET(instance, f"/orgs/{org}/teams")
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# COMMITS & HISTORY
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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,
|
|
}))
|
|
|
|
@tool
|
|
def get_commit(
|
|
owner: str, repo: str, sha: str,
|
|
stat: bool = True, files: bool = True, verification: bool = False,
|
|
instance: str = "",
|
|
) -> Any:
|
|
"""Get a single commit by SHA or ref. stat=True includes diff stats, files=True includes changed files list."""
|
|
return GET(instance, f"/repos/{owner}/{repo}/git/commits/{_b(sha)}", _strip({
|
|
"stat": stat, "files": files, "verification": verification or None,
|
|
}))
|
|
|
|
@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)}")
|
|
|
|
@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
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
|
|
@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}))
|
|
|
|
@tool
|
|
def mark_all_notifications_read(instance: str = "") -> Any:
|
|
"""Mark all notifications as read."""
|
|
return PUT(instance, "/notifications", {})
|
|
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
# CLI DISPATCHER (the second face of the same tools)
|
|
# ══════════════════════════════════════════════════════════════════════════════
|
|
#
|
|
# The same tool functions are exposed two ways from this one file:
|
|
# • MCP server — no subcommand → mcp.run() (for MCP-aware clients)
|
|
# • CLI — `<tool> --args` → one JSON line stdout (for Cline-style skills)
|
|
#
|
|
# CLI output contract: exactly ONE JSON object on stdout. Tool-level errors are
|
|
# JSON too ({"success": false, "error": ...}); the process prints and exits 0
|
|
# rather than crashing — the caller checks the `success` field, not the exit code.
|
|
# Hard startup failures (missing/invalid config) still exit non-zero, before this.
|
|
|
|
|
|
class _CliError(Exception):
|
|
"""Raised instead of argparse's SystemExit so we can emit a JSON envelope."""
|
|
|
|
|
|
class _ArgParser(argparse.ArgumentParser):
|
|
def error(self, message): # noqa: D102 — argparse override
|
|
raise _CliError(message)
|
|
|
|
|
|
def _arg_spec(annotation):
|
|
"""Map a parameter annotation to (kind, element_type) for argparse.
|
|
|
|
kind ∈ {"str", "int", "bool", "list"}. Unions like `bool | None` and
|
|
`list[str] | None` are unwrapped to their non-None inner type.
|
|
"""
|
|
origin = typing.get_origin(annotation)
|
|
if origin in (list, typing.List):
|
|
elem = (typing.get_args(annotation) or (str,))[0]
|
|
return "list", (int if elem is int else str)
|
|
if isinstance(annotation, type) and issubclass(annotation, bool):
|
|
return "bool", bool
|
|
if annotation is int:
|
|
return "int", int
|
|
if origin is typing.Union or isinstance(annotation, types.UnionType):
|
|
non_none = [a for a in typing.get_args(annotation) if a is not type(None)]
|
|
if non_none:
|
|
inner = non_none[0]
|
|
if inner is bool:
|
|
return "bool", bool
|
|
if typing.get_origin(inner) in (list, typing.List):
|
|
elem = (typing.get_args(inner) or (str,))[0]
|
|
return "list", (int if elem is int else str)
|
|
return "str", str
|
|
|
|
|
|
def _build_parser(name, fn):
|
|
"""Build an argparse parser for a tool from its signature."""
|
|
sig = inspect.signature(fn)
|
|
parser = _ArgParser(
|
|
prog=f"gitea {name}",
|
|
description=(fn.__doc__ or "").strip(),
|
|
add_help=False,
|
|
)
|
|
for pname, param in sig.parameters.items():
|
|
kind, etype = _arg_spec(param.annotation)
|
|
required = param.default is inspect.Parameter.empty
|
|
opt = "--" + pname.replace("_", "-")
|
|
if kind == "bool":
|
|
parser.add_argument(
|
|
opt, dest=pname, action=argparse.BooleanOptionalAction,
|
|
default=None, required=required,
|
|
)
|
|
elif kind == "list":
|
|
parser.add_argument(opt, dest=pname, nargs="*", type=etype, default=None, required=required)
|
|
elif kind == "int":
|
|
parser.add_argument(opt, dest=pname, type=int, default=None, required=required)
|
|
else:
|
|
parser.add_argument(opt, dest=pname, type=str, default=None, required=required)
|
|
return parser, sig
|
|
|
|
|
|
def _tool_list_payload():
|
|
"""Self-describing catalogue of every tool — for skill/agent discovery."""
|
|
tools = []
|
|
for name, fn in sorted(TOOLS.items()):
|
|
params = []
|
|
for pname, param in inspect.signature(fn).parameters.items():
|
|
kind, _ = _arg_spec(param.annotation)
|
|
params.append({
|
|
"name": pname,
|
|
"type": kind,
|
|
"required": param.default is inspect.Parameter.empty,
|
|
})
|
|
tools.append({
|
|
"name": name,
|
|
"description": (fn.__doc__ or "").strip(),
|
|
"params": params,
|
|
})
|
|
return {"success": True, "data": {"count": len(tools), "tools": tools}}
|
|
|
|
|
|
def dispatch(argv):
|
|
"""Run one CLI subcommand.
|
|
|
|
Returns an envelope dict to print, or None to signal "start the MCP server".
|
|
"""
|
|
if not argv or argv[0] in ("serve", "--serve"):
|
|
return None
|
|
cmd, rest = argv[0], argv[1:]
|
|
if cmd in ("list-tools", "--list-tools", "tools"):
|
|
return _tool_list_payload()
|
|
if cmd not in TOOLS:
|
|
return {
|
|
"success": False,
|
|
"error": f"Unknown command '{cmd}'. Run 'list-tools' to see all {len(TOOLS)} tools.",
|
|
"type": "UnknownCommand",
|
|
}
|
|
fn = TOOLS[cmd]
|
|
try:
|
|
parser, sig = _build_parser(cmd, fn)
|
|
ns = parser.parse_args(rest)
|
|
kwargs = {p: getattr(ns, p) for p in sig.parameters if getattr(ns, p) is not None}
|
|
return {"success": True, "data": fn(**kwargs)}
|
|
except _CliError as e:
|
|
return {"success": False, "error": str(e), "type": "UsageError"}
|
|
except Exception as e: # tool-level error → JSON, not a crash
|
|
return {"success": False, "error": str(e), "type": type(e).__name__}
|
|
|
|
|
|
def _drop_config(argv):
|
|
"""Strip --config X / --config=X (already consumed by find_config) from argv."""
|
|
out, skip = [], False
|
|
for a in argv:
|
|
if skip:
|
|
skip = False
|
|
continue
|
|
if a == "--config":
|
|
skip = True
|
|
continue
|
|
if a.startswith("--config="):
|
|
continue
|
|
out.append(a)
|
|
return out
|
|
|
|
|
|
def main():
|
|
envelope = dispatch(_drop_config(sys.argv[1:]))
|
|
if envelope is None:
|
|
mcp.run()
|
|
return
|
|
print(json.dumps(envelope, ensure_ascii=False, default=str))
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|