Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat(mcp): support native commit payload preparation without shell (closes #263)

Browse Source
This commit is contained in:
sysadmin
2026-07-06 14:42:59 -04:00
parent 67634c399e
commit 2a544b78d1
2 changed files with 358 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+107 -4
View File
@@ -2275,6 +2275,106 @@ def gitea_get_file(
}
def _prepare_commit_payload_files(files: list[dict]) -> tuple[list[dict], list[dict]]:
import base64
import json
import os
processed_files = []
source_proofs = []
lock_data = {}
if os.path.exists(ISSUE_LOCK_FILE):
try:
with open(ISSUE_LOCK_FILE, "r", encoding="utf-8") as f:
lock_data = json.load(f)
except Exception:
pass
locked_worktree = lock_data.get("worktree_path")
if locked_worktree:
locked_worktree = os.path.realpath(locked_worktree)
for f in files:
f_copy = dict(f)
path = f_copy.get("path", "")
content_keys = [k for k in ["content", "content_plain", "workspace_path", "local_path"] if k in f_copy]
if len(content_keys) > 1:
raise ValueError(
f"Multiple content sources specified for file '{path}'; provide exactly one of "
f"'content', 'content_plain', 'workspace_path', or 'local_path' (fail closed)"
)
source = "none"
if "content_plain" in f_copy:
content_plain = f_copy.pop("content_plain")
if content_plain is not None:
content_bytes = content_plain.encode("utf-8")
f_copy["content"] = base64.b64encode(content_bytes).decode("utf-8")
source = "inline_plain"
elif "workspace_path" in f_copy:
workspace_path = f_copy.pop("workspace_path")
if not locked_worktree:
raise RuntimeError(
f"Issue lock is missing or does not define a worktree path. "
f"Cannot resolve workspace_path '{workspace_path}' (fail closed)."
)
target_path = os.path.realpath(os.path.abspath(os.path.join(locked_worktree, workspace_path)))
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
if not (is_inside or is_tmp):
raise ValueError(
f"workspace_path '{workspace_path}' resolves to '{target_path}' which falls outside "
f"of locked worktree '{locked_worktree}' (fail closed)"
)
if not os.path.exists(target_path):
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
with open(target_path, "rb") as fh:
file_bytes = fh.read()
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
source = "workspace_path"
elif "local_path" in f_copy:
local_path = f_copy.pop("local_path")
if not locked_worktree:
raise RuntimeError(
f"Issue lock is missing or does not define a worktree path. "
f"Cannot resolve local_path '{local_path}' (fail closed)."
)
target_path = os.path.realpath(os.path.abspath(local_path))
is_inside = target_path == locked_worktree or target_path.startswith(os.path.join(locked_worktree, ""))
is_tmp = target_path.startswith(os.path.realpath("/tmp") + os.sep) or target_path.startswith("/tmp" + os.sep)
if not (is_inside or is_tmp):
raise ValueError(
f"local_path '{local_path}' resolves to '{target_path}' which falls outside "
f"of locked worktree '{locked_worktree}' (fail closed)"
)
if not os.path.exists(target_path):
raise FileNotFoundError(f"File not found: {target_path} (fail closed)")
with open(target_path, "rb") as fh:
file_bytes = fh.read()
f_copy["content"] = base64.b64encode(file_bytes).decode("utf-8")
source = "local_path"
elif "content" in f_copy:
source = "inline_base64"
processed_files.append(f_copy)
source_proofs.append({
"path": path,
"source": source
})
return processed_files, source_proofs
@mcp.tool()
def gitea_commit_files(
files: list[dict],
@@ -2289,7 +2389,7 @@ def gitea_commit_files(
"""Commit changes to multiple files in a Gitea repository in a single atomic commit.
Args:
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and 'content' (base64 encoded for create/update), and optionally 'sha' (required for update/delete) or 'from_path' (for rename).
files: List of file operations. Each file dict must contain 'operation' ('create', 'update', 'delete', 'rename'), 'path', and one of content payload sources: 'content' (base64), 'content_plain', 'workspace_path', or 'local_path'.
message: The commit message.
branch: Optional existing branch to start/commit from.
new_branch: Optional new branch name to create for this commit.
@@ -2325,12 +2425,14 @@ def gitea_commit_files(
return blocked
verify_preflight_purity(remote)
processed_files, source_proofs = _prepare_commit_payload_files(files)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
url = f"{repo_api_url(h, o, r)}/contents"
payload = {
"files": files,
"files": processed_files,
"message": message,
}
if branch is not None:
@@ -2341,13 +2443,14 @@ def gitea_commit_files(
with _audited("commit_files", host=h, remote=remote, org=o, repo=r,
target_branch=(new_branch or branch),
request_metadata={"message": message,
"paths": [f.get("path") for f in files],
"operations": [f.get("operation") for f in files]}):
"paths": [f.get("path") for f in processed_files],
"operations": [f.get("operation") for f in processed_files]}):
data = api_request("POST", url, auth, payload)
return {
"success": True,
"commit": data.get("commit", {}).get("sha", ""),
"branch": data.get("branch", {}).get("name", ""),
"content_source_proof": source_proofs,
}