refactor: make gitea a self-contained Cline skill
Relocate the script into the skill (skills/gitea/scripts/gitea_mcp.py) and add a PEP 723 dependency block, so `uv run scripts/gitea_mcp.py` provisions deps with no install step and the skill folder is portable (drop into ~/.cline/skills/). Drop the non-standard `metadata` frontmatter — Cline only recognizes name + description. - .mcp.json points at the relocated script and runs with --no-project (PEP 723) - SKILL.md uses relative scripts/ paths; metadata block removed - find_config walks up the tree to locate config.json (skill root or plugin root) - tests import gitea_mcp via pytest pythonpath; README/CLAUDE paths updated Tests: 119 passed, 92% coverage. Standalone list-tools verified via uv --no-project.
This commit is contained in:
+19
-21
@@ -8,8 +8,6 @@ description: >
|
||||
"переключись на work gitea", "gitea", "list repos", "create issue",
|
||||
"open PR", "merge pull request", "create release", "add webhook",
|
||||
"list branches", "gitea notifications", "check gitea connection".
|
||||
metadata:
|
||||
version: "0.3.0"
|
||||
---
|
||||
|
||||
# Gitea Platform Assistant
|
||||
@@ -18,9 +16,9 @@ Full Gitea REST API — 66 tools for repos, issues, PRs, branches, files, releas
|
||||
webhooks, organizations. One source file, two ways to call it (pick whichever your
|
||||
runtime gives you):
|
||||
|
||||
- **CLI (this skill):** `uv run … server/gitea_mcp.py <tool> --param value` →
|
||||
- **CLI (this skill):** `uv run scripts/gitea_mcp.py <tool> --param value` →
|
||||
prints **one JSON object** on stdout. This is the default for Cline.
|
||||
- **MCP server:** the same tools exposed as `mcp__gitea__<tool>` when the file is
|
||||
- **MCP server:** the same tools exposed as `mcp__gitea__<tool>` when the script is
|
||||
registered via `.mcp.json`. Use these instead of `execute_command` if the tools
|
||||
are already available to you.
|
||||
|
||||
@@ -29,21 +27,21 @@ differs (CLI uses `--dashed-flags`, MCP uses named arguments).
|
||||
|
||||
## Calling via CLI
|
||||
|
||||
```
|
||||
ROOT = <plugin folder> # the gitea-platform-mcp directory
|
||||
SCRIPT = $ROOT/server/gitea_mcp.py
|
||||
```
|
||||
|
||||
Run through `uv` (deps come from the project; no separate install):
|
||||
The script is bundled inside this skill at `scripts/gitea_mcp.py` and carries its
|
||||
own PEP 723 dependency block, so `uv` provisions everything on first run — no
|
||||
install step. Run it from the skill directory:
|
||||
|
||||
```
|
||||
uv run --project "$ROOT" "$SCRIPT" --config "$ROOT/config.json" <tool> [--flags]
|
||||
uv run scripts/gitea_mcp.py <tool> [--flags]
|
||||
```
|
||||
|
||||
Convert snake_case params to `--dashed` flags. Booleans use `--flag` / `--no-flag`
|
||||
(e.g. `--private` / `--no-private`). List params take space-separated values
|
||||
(e.g. `--topics ci backend`).
|
||||
|
||||
Config is auto-discovered (`config.json` in the skill folder or a parent). Pass
|
||||
`--config <path>` to point elsewhere.
|
||||
|
||||
**Output contract:** exactly one JSON object on stdout. Always check `success`:
|
||||
|
||||
```json
|
||||
@@ -56,7 +54,7 @@ The process exits 0 even on tool-level errors — read `success`, not the exit c
|
||||
## Discovery
|
||||
|
||||
```
|
||||
uv run --project "$ROOT" "$SCRIPT" list-tools
|
||||
uv run scripts/gitea_mcp.py list-tools
|
||||
```
|
||||
|
||||
Returns every tool with its description and parameter spec (name / type /
|
||||
@@ -67,12 +65,12 @@ required). Use it when unsure of a flag instead of guessing.
|
||||
Always start by checking the connection:
|
||||
|
||||
```
|
||||
uv run … "$SCRIPT" list_instances → configured instances + default
|
||||
uv run … "$SCRIPT" get_current_user → verifies auth works
|
||||
uv run scripts/gitea_mcp.py list_instances → configured instances + default
|
||||
uv run scripts/gitea_mcp.py get_current_user → verifies auth works
|
||||
```
|
||||
|
||||
If these fail with a config error, tell the user to fill in `config.json` in the
|
||||
plugin folder (copy `config.example.json`).
|
||||
If these fail with a config error, tell the user to create `config.json` in the
|
||||
skill folder (copy `config.example.json` from the plugin root).
|
||||
|
||||
## Instance Selection
|
||||
|
||||
@@ -158,18 +156,18 @@ Every tool takes an optional `--instance` (default `""` = use config default).
|
||||
|
||||
```
|
||||
# Open an issue
|
||||
uv run --project "$ROOT" "$SCRIPT" create_issue \
|
||||
uv run scripts/gitea_mcp.py create_issue \
|
||||
--owner andrey --repo infra --title "CI flaky on main" --body "Fails ~1/5 runs"
|
||||
|
||||
# Edit then commit a file (two steps — get sha first)
|
||||
uv run … "$SCRIPT" get_file --owner andrey --repo infra --filepath README.md
|
||||
uv run … "$SCRIPT" update_file --owner andrey --repo infra --filepath README.md \
|
||||
uv run scripts/gitea_mcp.py get_file --owner andrey --repo infra --filepath README.md
|
||||
uv run scripts/gitea_mcp.py update_file --owner andrey --repo infra --filepath README.md \
|
||||
--content "new text" --message "docs: tweak" --sha <sha-from-get_file>
|
||||
|
||||
# Create and merge a PR on the work instance
|
||||
uv run … "$SCRIPT" create_pull_request --owner team --repo svc \
|
||||
uv run scripts/gitea_mcp.py create_pull_request --owner team --repo svc \
|
||||
--title "Feature X" --head feature/x --base main --instance work
|
||||
uv run … "$SCRIPT" merge_pull_request --owner team --repo svc --index 42 \
|
||||
uv run scripts/gitea_mcp.py merge_pull_request --owner team --repo svc --index 42 \
|
||||
--merge-style squash --delete-branch-after-merge --instance work
|
||||
```
|
||||
|
||||
|
||||
@@ -0,0 +1,856 @@
|
||||
#!/usr/bin/env python3
|
||||
# /// script
|
||||
# requires-python = ">=3.10"
|
||||
# dependencies = ["mcp[cli]>=1.0", "httpx>=0.27"]
|
||||
# ///
|
||||
"""
|
||||
Gitea platform tools — full Gitea management via the REST API.
|
||||
|
||||
Dual-mode, one file:
|
||||
• MCP server: no subcommand -> mcp.run() (registered via .mcp.json)
|
||||
• CLI skill: `<tool> --flags` -> one JSON line (run via `uv run`)
|
||||
|
||||
The PEP 723 block above lets `uv run gitea_mcp.py ...` provision deps with no
|
||||
separate install, so the skill folder is self-contained.
|
||||
|
||||
Reads credentials from config.json (--config argument, GITEA_CONFIG env, or
|
||||
auto-discovered by walking up from the script).
|
||||
"""
|
||||
|
||||
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 > walk up from script > cwd.
|
||||
|
||||
The script lives at skills/gitea/scripts/, so config.json may sit in the skill
|
||||
folder (skills/gitea/) or at the plugin root — walk up a few levels to find it.
|
||||
"""
|
||||
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"]
|
||||
# Walk up from the script directory looking for config.json (skill root,
|
||||
# then plugin root, etc.).
|
||||
here = os.path.dirname(os.path.abspath(__file__))
|
||||
for _ in range(5):
|
||||
candidate = os.path.join(here, "config.json")
|
||||
if os.path.exists(candidate):
|
||||
return candidate
|
||||
here = os.path.dirname(here)
|
||||
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()
|
||||
Reference in New Issue
Block a user