Merge pull request 'Feat/expand gitea toolset' (#20) from feat/expand-gitea-toolset into main
Reviewed-on: #20
This commit was merged in pull request #20.
This commit is contained in:
+29
-1
@@ -58,6 +58,19 @@ Scope requirements:
|
||||
- `list_labels` (`owner`, `repo`, optional `page`, `limit`)
|
||||
- `list_tags` (`owner`, `repo`, optional `page`, `limit`)
|
||||
- `list_releases` (`owner`, `repo`, optional `page`, `limit`)
|
||||
- `list_pull_request_files` (`owner`, `repo`, `pull_number`, optional `page`, `limit`)
|
||||
- `list_pull_request_commits` (`owner`, `repo`, `pull_number`, optional `page`, `limit`)
|
||||
- `list_issue_comments` (`owner`, `repo`, `issue_number`, optional `page`, `limit`)
|
||||
- `list_branches` (`owner`, `repo`, optional `page`, `limit`)
|
||||
- `get_branch` (`owner`, `repo`, `branch`)
|
||||
- `get_release` (`owner`, `repo`, `release_id`)
|
||||
- `get_latest_release` (`owner`, `repo`)
|
||||
- `list_milestones` (`owner`, `repo`, optional `state`, `page`, `limit`)
|
||||
- `get_commit_status` (`owner`, `repo`, `sha`)
|
||||
- `list_org_repositories` (`org`, optional `page`, `limit`)
|
||||
- `list_organizations` (optional `page`, `limit`)
|
||||
- `get_repo_languages` (`owner`, `repo`)
|
||||
- `list_repo_topics` (`owner`, `repo`)
|
||||
|
||||
## Write Tools (Write Mode Required)
|
||||
|
||||
@@ -65,8 +78,23 @@ Scope requirements:
|
||||
- `update_issue` (`owner`, `repo`, `issue_number`, one or more of `title`, `body`, `state`)
|
||||
- `create_issue_comment` (`owner`, `repo`, `issue_number`, `body`)
|
||||
- `create_pr_comment` (`owner`, `repo`, `pull_number`, `body`)
|
||||
- `add_labels` (`owner`, `repo`, `issue_number`, `labels`)
|
||||
- `add_labels` (`owner`, `repo`, `issue_number`, `labels` by name)
|
||||
- `remove_labels` (`owner`, `repo`, `issue_number`, `labels` by name)
|
||||
- `assign_issue` (`owner`, `repo`, `issue_number`, `assignees`)
|
||||
- `create_label` (`owner`, `repo`, `name`, `color` hex e.g. `#00aabb`, optional `description`, `exclusive`)
|
||||
- `update_label` (`owner`, `repo`, `name`, one or more of `new_name`, `color`, `description`)
|
||||
- `create_pull_request` (`owner`, `repo`, `title`, `head`, `base`, optional `body`)
|
||||
- `create_release` (`owner`, `repo`, `tag_name`, optional `name`, `body`, `draft`, `prerelease`, `target`)
|
||||
- `edit_release` (`owner`, `repo`, `release_id`, one or more of `name`, `body`, `draft`, `prerelease`)
|
||||
- `create_branch` (`owner`, `repo`, `new_branch_name`, optional `old_branch_name`)
|
||||
- `create_milestone` (`owner`, `repo`, `title`, optional `description`, `due_on`)
|
||||
- `edit_issue_comment` (`owner`, `repo`, `comment_id`, `body`)
|
||||
|
||||
Not supported by design: merge, branch/label/release deletion, force push, repo/admin
|
||||
management.
|
||||
|
||||
Note: `create_issue`, `add_labels`, and `remove_labels` accept label **names**; the
|
||||
server resolves them to Gitea label ids and returns a clear error for unknown labels.
|
||||
|
||||
## Validation and Limits
|
||||
|
||||
|
||||
@@ -582,6 +582,45 @@ class GiteaClient:
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def _resolve_label_ids(
|
||||
self, owner: str, repo: str, names: list[str], *, correlation_id: str
|
||||
) -> list[int]:
|
||||
"""Resolve label names to Gitea label ids for a repository.
|
||||
|
||||
Gitea's issue/label APIs require numeric label ids, not names. This maps
|
||||
the caller-supplied names (case-insensitive) to ids and raises a clear
|
||||
error for any name that does not exist in the repository.
|
||||
"""
|
||||
existing = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/labels",
|
||||
params={"limit": 100},
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
by_name: dict[str, int] = {}
|
||||
if isinstance(existing, list):
|
||||
for item in existing:
|
||||
if isinstance(item, dict):
|
||||
label_name = str(item.get("name", ""))
|
||||
label_id = item.get("id")
|
||||
if label_name and isinstance(label_id, int):
|
||||
by_name[label_name.lower()] = label_id
|
||||
|
||||
ids: list[int] = []
|
||||
unknown: list[str] = []
|
||||
for name in names:
|
||||
match = by_name.get(name.strip().lower())
|
||||
if match is None:
|
||||
unknown.append(name)
|
||||
else:
|
||||
ids.append(match)
|
||||
if unknown:
|
||||
raise GiteaError(
|
||||
f"Unknown label(s) for {owner}/{repo}: {', '.join(unknown)}. "
|
||||
"Create them first with create_label."
|
||||
)
|
||||
return ids
|
||||
|
||||
async def create_issue(
|
||||
self,
|
||||
owner: str,
|
||||
@@ -593,18 +632,21 @@ class GiteaClient:
|
||||
assignees: list[str] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create repository issue."""
|
||||
correlation_id = str(
|
||||
self.audit.log_tool_invocation(tool_name="create_issue", result_status="pending")
|
||||
)
|
||||
payload: dict[str, Any] = {"title": title, "body": body}
|
||||
if labels:
|
||||
payload["labels"] = labels
|
||||
payload["labels"] = await self._resolve_label_ids(
|
||||
owner, repo, labels, correlation_id=correlation_id
|
||||
)
|
||||
if assignees:
|
||||
payload["assignees"] = assignees
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/issues",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="create_issue", result_status="pending")
|
||||
),
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
@@ -668,6 +710,33 @@ class GiteaClient:
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def create_label(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
name: str,
|
||||
color: str,
|
||||
description: str = "",
|
||||
exclusive: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a repository label."""
|
||||
payload: dict[str, Any] = {
|
||||
"name": name,
|
||||
"color": color,
|
||||
"description": description,
|
||||
"exclusive": exclusive,
|
||||
}
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/labels",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="create_label", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def add_labels(
|
||||
self,
|
||||
owner: str,
|
||||
@@ -675,14 +744,82 @@ class GiteaClient:
|
||||
index: int,
|
||||
labels: list[str],
|
||||
) -> dict[str, Any]:
|
||||
"""Add labels to issue/PR."""
|
||||
"""Add labels to issue/PR by label name."""
|
||||
correlation_id = str(
|
||||
self.audit.log_tool_invocation(tool_name="add_labels", result_status="pending")
|
||||
)
|
||||
label_ids = await self._resolve_label_ids(
|
||||
owner, repo, labels, correlation_id=correlation_id
|
||||
)
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/issues/{index}/labels",
|
||||
json_body={"labels": labels},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="add_labels", result_status="pending")
|
||||
),
|
||||
json_body={"labels": label_ids},
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def remove_labels(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
index: int,
|
||||
labels: list[str],
|
||||
) -> list[dict[str, Any]]:
|
||||
"""Remove the given labels (by name) from an issue/PR.
|
||||
|
||||
Returns the issue's remaining labels.
|
||||
"""
|
||||
correlation_id = str(
|
||||
self.audit.log_tool_invocation(tool_name="remove_labels", result_status="pending")
|
||||
)
|
||||
label_ids = await self._resolve_label_ids(
|
||||
owner, repo, labels, correlation_id=correlation_id
|
||||
)
|
||||
owner_q = quote(owner, safe="")
|
||||
repo_q = quote(repo, safe="")
|
||||
for label_id in label_ids:
|
||||
await self._request(
|
||||
"DELETE",
|
||||
f"/api/v1/repos/{owner_q}/{repo_q}/issues/{index}/labels/{label_id}",
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{owner_q}/{repo_q}/issues/{index}/labels",
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def update_label(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
name: str,
|
||||
new_name: str | None = None,
|
||||
color: str | None = None,
|
||||
description: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Update an existing label, located by its current name."""
|
||||
correlation_id = str(
|
||||
self.audit.log_tool_invocation(tool_name="update_label", result_status="pending")
|
||||
)
|
||||
label_ids = await self._resolve_label_ids(
|
||||
owner, repo, [name], correlation_id=correlation_id
|
||||
)
|
||||
payload: dict[str, Any] = {}
|
||||
if new_name is not None:
|
||||
payload["name"] = new_name
|
||||
if color is not None:
|
||||
payload["color"] = color
|
||||
if description is not None:
|
||||
payload["description"] = description
|
||||
result = await self._request(
|
||||
"PATCH",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/labels/{label_ids[0]}",
|
||||
json_body=payload,
|
||||
correlation_id=correlation_id,
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
@@ -703,3 +840,335 @@ class GiteaClient:
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def create_pull_request(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
title: str,
|
||||
head: str,
|
||||
base: str,
|
||||
body: str = "",
|
||||
) -> dict[str, Any]:
|
||||
"""Open a pull request from head into base."""
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/pulls",
|
||||
json_body={"title": title, "head": head, "base": base, "body": body},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="create_pull_request", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def create_release(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
tag_name: str,
|
||||
name: str = "",
|
||||
body: str = "",
|
||||
draft: bool = False,
|
||||
prerelease: bool = False,
|
||||
target: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a release for an existing or new tag."""
|
||||
payload: dict[str, Any] = {
|
||||
"tag_name": tag_name,
|
||||
"name": name or tag_name,
|
||||
"body": body,
|
||||
"draft": draft,
|
||||
"prerelease": prerelease,
|
||||
}
|
||||
if target:
|
||||
payload["target_commitish"] = target
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/releases",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="create_release", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def edit_release(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
release_id: int,
|
||||
*,
|
||||
name: str | None = None,
|
||||
body: str | None = None,
|
||||
draft: bool | None = None,
|
||||
prerelease: bool | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Edit fields of an existing release."""
|
||||
payload: dict[str, Any] = {}
|
||||
if name is not None:
|
||||
payload["name"] = name
|
||||
if body is not None:
|
||||
payload["body"] = body
|
||||
if draft is not None:
|
||||
payload["draft"] = draft
|
||||
if prerelease is not None:
|
||||
payload["prerelease"] = prerelease
|
||||
result = await self._request(
|
||||
"PATCH",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/releases/{release_id}",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="edit_release", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def create_branch(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
new_branch_name: str,
|
||||
old_branch_name: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a branch, optionally from a specific existing branch."""
|
||||
payload: dict[str, Any] = {"new_branch_name": new_branch_name}
|
||||
if old_branch_name:
|
||||
payload["old_branch_name"] = old_branch_name
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="create_branch", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def create_milestone(
|
||||
self,
|
||||
owner: str,
|
||||
repo: str,
|
||||
*,
|
||||
title: str,
|
||||
description: str = "",
|
||||
due_on: str | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a repository milestone."""
|
||||
payload: dict[str, Any] = {"title": title, "description": description}
|
||||
if due_on:
|
||||
payload["due_on"] = due_on
|
||||
result = await self._request(
|
||||
"POST",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/milestones",
|
||||
json_body=payload,
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="create_milestone", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def edit_issue_comment(
|
||||
self, owner: str, repo: str, comment_id: int, body: str
|
||||
) -> dict[str, Any]:
|
||||
"""Edit an existing issue or PR comment."""
|
||||
result = await self._request(
|
||||
"PATCH",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/issues/comments/{comment_id}",
|
||||
json_body={"body": body},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="edit_issue_comment", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def list_pull_request_files(
|
||||
self, owner: str, repo: str, index: int, *, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List files changed in a pull request."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/pulls/{index}/files",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_pull_request_files", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def list_pull_request_commits(
|
||||
self, owner: str, repo: str, index: int, *, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List commits in a pull request."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/pulls/{index}/commits",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_pull_request_commits", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def list_issue_comments(
|
||||
self, owner: str, repo: str, index: int, *, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List comments on an issue or pull request."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/issues/{index}/comments",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_issue_comments", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def list_branches(
|
||||
self, owner: str, repo: str, *, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List repository branches."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="list_branches", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_branch(self, owner: str, repo: str, branch: str) -> dict[str, Any]:
|
||||
"""Get a single branch."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/branches/{quote(branch, safe='/')}",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="get_branch", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def get_release(self, owner: str, repo: str, release_id: int) -> dict[str, Any]:
|
||||
"""Get a release by id."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/releases/{release_id}",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="get_release", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def get_latest_release(self, owner: str, repo: str) -> dict[str, Any]:
|
||||
"""Get the latest published release."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/releases/latest",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="get_latest_release", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def list_milestones(
|
||||
self, owner: str, repo: str, *, state: str, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List repository milestones."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/milestones",
|
||||
params={"state": state, "page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(tool_name="list_milestones", result_status="pending")
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_commit_status(self, owner: str, repo: str, sha: str) -> dict[str, Any]:
|
||||
"""Get the combined commit status for a ref/sha."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/commits/{quote(sha, safe='/')}/status",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="get_commit_status", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def list_org_repositories(
|
||||
self, org: str, *, page: int, limit: int
|
||||
) -> list[dict[str, Any]]:
|
||||
"""List repositories belonging to an organization."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/orgs/{quote(org, safe='')}/repos",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_org_repositories", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def list_organizations(self, *, page: int, limit: int) -> list[dict[str, Any]]:
|
||||
"""List organizations the authenticated user belongs to."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
"/api/v1/user/orgs",
|
||||
params={"page": page, "limit": limit},
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_organizations", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, list) else []
|
||||
|
||||
async def get_repo_languages(self, owner: str, repo: str) -> dict[str, Any]:
|
||||
"""Get the language breakdown for a repository."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/languages",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="get_repo_languages", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
return result if isinstance(result, dict) else {}
|
||||
|
||||
async def list_repo_topics(self, owner: str, repo: str) -> list[str]:
|
||||
"""List the topics assigned to a repository."""
|
||||
result = await self._request(
|
||||
"GET",
|
||||
f"/api/v1/repos/{quote(owner, safe='')}/{quote(repo, safe='')}/topics",
|
||||
correlation_id=str(
|
||||
self.audit.log_tool_invocation(
|
||||
tool_name="list_repo_topics", result_status="pending"
|
||||
)
|
||||
),
|
||||
)
|
||||
if isinstance(result, dict):
|
||||
topics = result.get("topics", [])
|
||||
return [str(topic) for topic in topics] if isinstance(topics, list) else []
|
||||
return []
|
||||
|
||||
@@ -274,6 +274,184 @@ AVAILABLE_TOOLS: list[MCPTool] = [
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_pull_request_files",
|
||||
"List files changed in a pull request.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"pull_number": {"type": "integer", "minimum": 1},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["owner", "repo", "pull_number"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_pull_request_commits",
|
||||
"List commits in a pull request.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"pull_number": {"type": "integer", "minimum": 1},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["owner", "repo", "pull_number"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_issue_comments",
|
||||
"List comments on an issue or pull request.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"issue_number": {"type": "integer", "minimum": 1},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["owner", "repo", "issue_number"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_branches",
|
||||
"List repository branches.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["owner", "repo"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"get_branch",
|
||||
"Get a single branch.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"branch": {"type": "string"},
|
||||
},
|
||||
"required": ["owner", "repo", "branch"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"get_release",
|
||||
"Get a release by id.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"release_id": {"type": "integer", "minimum": 1},
|
||||
},
|
||||
"required": ["owner", "repo", "release_id"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"get_latest_release",
|
||||
"Get the latest published release.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"owner": {"type": "string"}, "repo": {"type": "string"}},
|
||||
"required": ["owner", "repo"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_milestones",
|
||||
"List repository milestones.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"state": {"type": "string", "enum": ["open", "closed", "all"], "default": "open"},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["owner", "repo"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"get_commit_status",
|
||||
"Get the combined commit status for a ref or sha.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"sha": {"type": "string"},
|
||||
},
|
||||
"required": ["owner", "repo", "sha"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_org_repositories",
|
||||
"List repositories belonging to an organization.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"org": {"type": "string"},
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": ["org"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_organizations",
|
||||
"List organizations the authenticated user belongs to.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"page": {"type": "integer", "minimum": 1, "default": 1},
|
||||
"limit": {"type": "integer", "minimum": 1, "maximum": 100, "default": 50},
|
||||
},
|
||||
"required": [],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"get_repo_languages",
|
||||
"Get the language breakdown for a repository.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"owner": {"type": "string"}, "repo": {"type": "string"}},
|
||||
"required": ["owner", "repo"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"list_repo_topics",
|
||||
"List the topics assigned to a repository.",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {"owner": {"type": "string"}, "repo": {"type": "string"}},
|
||||
"required": ["owner", "repo"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
),
|
||||
_tool(
|
||||
"create_issue",
|
||||
"Create a repository issue (write-mode only).",
|
||||
@@ -374,6 +552,164 @@ AVAILABLE_TOOLS: list[MCPTool] = [
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"create_label",
|
||||
"Create a repository label (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"name": {"type": "string"},
|
||||
"color": {"type": "string", "description": "Hex color, e.g. #00aabb"},
|
||||
"description": {"type": "string", "default": ""},
|
||||
"exclusive": {"type": "boolean", "default": False},
|
||||
},
|
||||
"required": ["owner", "repo", "name", "color"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"update_label",
|
||||
"Update an existing repository label, located by its current name (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"name": {"type": "string", "description": "Current label name"},
|
||||
"new_name": {"type": "string"},
|
||||
"color": {"type": "string", "description": "Hex color, e.g. #00aabb"},
|
||||
"description": {"type": "string"},
|
||||
},
|
||||
"required": ["owner", "repo", "name"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"remove_labels",
|
||||
"Remove labels (by name) from an issue or pull request (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"issue_number": {"type": "integer", "minimum": 1},
|
||||
"labels": {"type": "array", "items": {"type": "string"}, "minItems": 1},
|
||||
},
|
||||
"required": ["owner", "repo", "issue_number", "labels"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"create_pull_request",
|
||||
"Open a pull request from head into base (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"title": {"type": "string"},
|
||||
"head": {"type": "string", "description": "Source branch"},
|
||||
"base": {"type": "string", "description": "Target branch"},
|
||||
"body": {"type": "string", "default": ""},
|
||||
},
|
||||
"required": ["owner", "repo", "title", "head", "base"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"create_release",
|
||||
"Create a release for a tag (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"tag_name": {"type": "string"},
|
||||
"name": {"type": "string", "default": ""},
|
||||
"body": {"type": "string", "default": ""},
|
||||
"draft": {"type": "boolean", "default": False},
|
||||
"prerelease": {"type": "boolean", "default": False},
|
||||
"target": {"type": "string", "description": "Target commitish/branch"},
|
||||
},
|
||||
"required": ["owner", "repo", "tag_name"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"edit_release",
|
||||
"Edit an existing release (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"release_id": {"type": "integer", "minimum": 1},
|
||||
"name": {"type": "string"},
|
||||
"body": {"type": "string"},
|
||||
"draft": {"type": "boolean"},
|
||||
"prerelease": {"type": "boolean"},
|
||||
},
|
||||
"required": ["owner", "repo", "release_id"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"create_branch",
|
||||
"Create a branch, optionally from an existing branch (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"new_branch_name": {"type": "string"},
|
||||
"old_branch_name": {"type": "string", "description": "Source branch (optional)"},
|
||||
},
|
||||
"required": ["owner", "repo", "new_branch_name"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"create_milestone",
|
||||
"Create a repository milestone (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"title": {"type": "string"},
|
||||
"description": {"type": "string", "default": ""},
|
||||
"due_on": {"type": "string", "description": "ISO8601 due date (optional)"},
|
||||
},
|
||||
"required": ["owner", "repo", "title"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
_tool(
|
||||
"edit_issue_comment",
|
||||
"Edit an existing issue or PR comment (write-mode only).",
|
||||
{
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"owner": {"type": "string"},
|
||||
"repo": {"type": "string"},
|
||||
"comment_id": {"type": "integer", "minimum": 1},
|
||||
"body": {"type": "string"},
|
||||
},
|
||||
"required": ["owner", "repo", "comment_id", "body"],
|
||||
"additionalProperties": False,
|
||||
},
|
||||
write_operation=True,
|
||||
),
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -61,14 +61,27 @@ from aegis_gitea_mcp.security import sanitize_data
|
||||
from aegis_gitea_mcp.tools.arguments import extract_repository, extract_target_path
|
||||
from aegis_gitea_mcp.tools.read_tools import (
|
||||
compare_refs_tool,
|
||||
get_branch_tool,
|
||||
get_commit_diff_tool,
|
||||
get_commit_status_tool,
|
||||
get_issue_tool,
|
||||
get_latest_release_tool,
|
||||
get_pull_request_tool,
|
||||
get_release_tool,
|
||||
get_repo_languages_tool,
|
||||
list_branches_tool,
|
||||
list_commits_tool,
|
||||
list_issue_comments_tool,
|
||||
list_issues_tool,
|
||||
list_labels_tool,
|
||||
list_milestones_tool,
|
||||
list_org_repositories_tool,
|
||||
list_organizations_tool,
|
||||
list_pull_request_commits_tool,
|
||||
list_pull_request_files_tool,
|
||||
list_pull_requests_tool,
|
||||
list_releases_tool,
|
||||
list_repo_topics_tool,
|
||||
list_tags_tool,
|
||||
search_code_tool,
|
||||
)
|
||||
@@ -81,10 +94,19 @@ from aegis_gitea_mcp.tools.repository import (
|
||||
from aegis_gitea_mcp.tools.write_tools import (
|
||||
add_labels_tool,
|
||||
assign_issue_tool,
|
||||
create_branch_tool,
|
||||
create_issue_comment_tool,
|
||||
create_issue_tool,
|
||||
create_label_tool,
|
||||
create_milestone_tool,
|
||||
create_pr_comment_tool,
|
||||
create_pull_request_tool,
|
||||
create_release_tool,
|
||||
edit_issue_comment_tool,
|
||||
edit_release_tool,
|
||||
remove_labels_tool,
|
||||
update_issue_tool,
|
||||
update_label_tool,
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -335,6 +357,19 @@ TOOL_HANDLERS: dict[str, ToolHandler] = {
|
||||
"list_labels": list_labels_tool,
|
||||
"list_tags": list_tags_tool,
|
||||
"list_releases": list_releases_tool,
|
||||
"list_pull_request_files": list_pull_request_files_tool,
|
||||
"list_pull_request_commits": list_pull_request_commits_tool,
|
||||
"list_issue_comments": list_issue_comments_tool,
|
||||
"list_branches": list_branches_tool,
|
||||
"get_branch": get_branch_tool,
|
||||
"get_release": get_release_tool,
|
||||
"get_latest_release": get_latest_release_tool,
|
||||
"list_milestones": list_milestones_tool,
|
||||
"get_commit_status": get_commit_status_tool,
|
||||
"list_org_repositories": list_org_repositories_tool,
|
||||
"list_organizations": list_organizations_tool,
|
||||
"get_repo_languages": get_repo_languages_tool,
|
||||
"list_repo_topics": list_repo_topics_tool,
|
||||
# Write-mode tools
|
||||
"create_issue": create_issue_tool,
|
||||
"update_issue": update_issue_tool,
|
||||
@@ -342,6 +377,15 @@ TOOL_HANDLERS: dict[str, ToolHandler] = {
|
||||
"create_pr_comment": create_pr_comment_tool,
|
||||
"add_labels": add_labels_tool,
|
||||
"assign_issue": assign_issue_tool,
|
||||
"create_label": create_label_tool,
|
||||
"update_label": update_label_tool,
|
||||
"remove_labels": remove_labels_tool,
|
||||
"create_pull_request": create_pull_request_tool,
|
||||
"create_release": create_release_tool,
|
||||
"edit_release": edit_release_tool,
|
||||
"create_branch": create_branch_tool,
|
||||
"create_milestone": create_milestone_tool,
|
||||
"edit_issue_comment": edit_issue_comment_tool,
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -220,6 +220,187 @@ class AssignIssueArgs(RepositoryArgs):
|
||||
assignees: list[str] = Field(..., min_length=1, max_length=20)
|
||||
|
||||
|
||||
class CreateLabelArgs(RepositoryArgs):
|
||||
"""Arguments for create_label."""
|
||||
|
||||
name: str = Field(..., min_length=1, max_length=50)
|
||||
# Gitea requires a hex color; accept it with or without a leading '#'.
|
||||
color: str = Field(..., pattern=r"^#?[0-9A-Fa-f]{6}$")
|
||||
description: str = Field(default="", max_length=1000)
|
||||
exclusive: bool = Field(default=False)
|
||||
|
||||
|
||||
class UpdateLabelArgs(RepositoryArgs):
|
||||
"""Arguments for update_label (located by current name)."""
|
||||
|
||||
name: str = Field(..., min_length=1, max_length=50)
|
||||
new_name: str | None = Field(default=None, min_length=1, max_length=50)
|
||||
color: str | None = Field(default=None, pattern=r"^#?[0-9A-Fa-f]{6}$")
|
||||
description: str | None = Field(default=None, max_length=1000)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def require_change(self) -> UpdateLabelArgs:
|
||||
"""Require at least one mutable field in the update payload."""
|
||||
if self.new_name is None and self.color is None and self.description is None:
|
||||
raise ValueError("At least one of new_name, color, or description must be provided")
|
||||
return self
|
||||
|
||||
|
||||
class RemoveLabelsArgs(RepositoryArgs):
|
||||
"""Arguments for remove_labels."""
|
||||
|
||||
issue_number: int = Field(..., ge=1)
|
||||
labels: list[str] = Field(..., min_length=1, max_length=20)
|
||||
|
||||
|
||||
class CreatePullRequestArgs(RepositoryArgs):
|
||||
"""Arguments for create_pull_request."""
|
||||
|
||||
title: str = Field(..., min_length=1, max_length=256)
|
||||
head: GitRef = Field(..., min_length=1, max_length=200)
|
||||
base: GitRef = Field(..., min_length=1, max_length=200)
|
||||
body: str = Field(default="", max_length=20_000)
|
||||
|
||||
|
||||
class CreateReleaseArgs(RepositoryArgs):
|
||||
"""Arguments for create_release."""
|
||||
|
||||
tag_name: GitRef = Field(..., min_length=1, max_length=200)
|
||||
name: str = Field(default="", max_length=256)
|
||||
body: str = Field(default="", max_length=20_000)
|
||||
draft: bool = Field(default=False)
|
||||
prerelease: bool = Field(default=False)
|
||||
target: str | None = Field(default=None, min_length=1, max_length=200)
|
||||
|
||||
|
||||
class EditReleaseArgs(RepositoryArgs):
|
||||
"""Arguments for edit_release."""
|
||||
|
||||
release_id: int = Field(..., ge=1)
|
||||
name: str | None = Field(default=None, max_length=256)
|
||||
body: str | None = Field(default=None, max_length=20_000)
|
||||
draft: bool | None = Field(default=None)
|
||||
prerelease: bool | None = Field(default=None)
|
||||
|
||||
@model_validator(mode="after")
|
||||
def require_change(self) -> EditReleaseArgs:
|
||||
"""Require at least one mutable field in the update payload."""
|
||||
if (
|
||||
self.name is None
|
||||
and self.body is None
|
||||
and self.draft is None
|
||||
and self.prerelease is None
|
||||
):
|
||||
raise ValueError("At least one of name, body, draft, or prerelease must be provided")
|
||||
return self
|
||||
|
||||
|
||||
class CreateBranchArgs(RepositoryArgs):
|
||||
"""Arguments for create_branch."""
|
||||
|
||||
new_branch_name: GitRef = Field(..., min_length=1, max_length=200)
|
||||
old_branch_name: str | None = Field(default=None, min_length=1, max_length=200)
|
||||
|
||||
|
||||
class CreateMilestoneArgs(RepositoryArgs):
|
||||
"""Arguments for create_milestone."""
|
||||
|
||||
title: str = Field(..., min_length=1, max_length=256)
|
||||
description: str = Field(default="", max_length=10_000)
|
||||
due_on: str | None = Field(default=None, max_length=64)
|
||||
|
||||
|
||||
class EditIssueCommentArgs(RepositoryArgs):
|
||||
"""Arguments for edit_issue_comment."""
|
||||
|
||||
comment_id: int = Field(..., ge=1)
|
||||
body: str = Field(..., min_length=1, max_length=10_000)
|
||||
|
||||
|
||||
class ListPullRequestFilesArgs(RepositoryArgs):
|
||||
"""Arguments for list_pull_request_files."""
|
||||
|
||||
pull_number: int = Field(..., ge=1)
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class ListPullRequestCommitsArgs(RepositoryArgs):
|
||||
"""Arguments for list_pull_request_commits."""
|
||||
|
||||
pull_number: int = Field(..., ge=1)
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class ListIssueCommentsArgs(RepositoryArgs):
|
||||
"""Arguments for list_issue_comments."""
|
||||
|
||||
issue_number: int = Field(..., ge=1)
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class ListBranchesArgs(RepositoryArgs):
|
||||
"""Arguments for list_branches."""
|
||||
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class GetBranchArgs(RepositoryArgs):
|
||||
"""Arguments for get_branch."""
|
||||
|
||||
branch: GitRef = Field(..., min_length=1, max_length=200)
|
||||
|
||||
|
||||
class GetReleaseArgs(RepositoryArgs):
|
||||
"""Arguments for get_release."""
|
||||
|
||||
release_id: int = Field(..., ge=1)
|
||||
|
||||
|
||||
class LatestReleaseArgs(RepositoryArgs):
|
||||
"""Arguments for get_latest_release."""
|
||||
|
||||
|
||||
class ListMilestonesArgs(RepositoryArgs):
|
||||
"""Arguments for list_milestones."""
|
||||
|
||||
state: Literal["open", "closed", "all"] = Field(default="open")
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class CommitStatusArgs(RepositoryArgs):
|
||||
"""Arguments for get_commit_status."""
|
||||
|
||||
sha: GitRef = Field(..., min_length=1, max_length=64)
|
||||
|
||||
|
||||
class ListOrgRepositoriesArgs(StrictBaseModel):
|
||||
"""Arguments for list_org_repositories."""
|
||||
|
||||
org: str = Field(..., pattern=_REPO_PART_PATTERN)
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class ListOrganizationsArgs(StrictBaseModel):
|
||||
"""Arguments for list_organizations."""
|
||||
|
||||
page: int = Field(default=1, ge=1, le=10_000)
|
||||
limit: int = Field(default=50, ge=1, le=100)
|
||||
|
||||
|
||||
class RepoLanguagesArgs(RepositoryArgs):
|
||||
"""Arguments for get_repo_languages."""
|
||||
|
||||
|
||||
class RepoTopicsArgs(RepositoryArgs):
|
||||
"""Arguments for list_repo_topics."""
|
||||
|
||||
|
||||
def extract_repository(arguments: dict[str, object]) -> str | None:
|
||||
"""Extract `owner/repo` from raw argument mapping.
|
||||
|
||||
|
||||
@@ -13,15 +13,28 @@ from aegis_gitea_mcp.gitea_client import (
|
||||
from aegis_gitea_mcp.response_limits import limit_items, limit_text
|
||||
from aegis_gitea_mcp.tools.arguments import (
|
||||
CommitDiffArgs,
|
||||
CommitStatusArgs,
|
||||
CompareRefsArgs,
|
||||
GetBranchArgs,
|
||||
GetReleaseArgs,
|
||||
IssueArgs,
|
||||
LatestReleaseArgs,
|
||||
ListBranchesArgs,
|
||||
ListCommitsArgs,
|
||||
ListIssueCommentsArgs,
|
||||
ListIssuesArgs,
|
||||
ListLabelsArgs,
|
||||
ListMilestonesArgs,
|
||||
ListOrganizationsArgs,
|
||||
ListOrgRepositoriesArgs,
|
||||
ListPullRequestCommitsArgs,
|
||||
ListPullRequestFilesArgs,
|
||||
ListPullRequestsArgs,
|
||||
ListReleasesArgs,
|
||||
ListTagsArgs,
|
||||
PullRequestArgs,
|
||||
RepoLanguagesArgs,
|
||||
RepoTopicsArgs,
|
||||
SearchCodeArgs,
|
||||
)
|
||||
|
||||
@@ -449,3 +462,359 @@ async def list_releases_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> d
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list releases: {exc}") from exc
|
||||
|
||||
|
||||
async def list_pull_request_files_tool(
|
||||
gitea: GiteaClient, arguments: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""List files changed in a pull request."""
|
||||
parsed = ListPullRequestFilesArgs.model_validate(arguments)
|
||||
try:
|
||||
files = await gitea.list_pull_request_files(
|
||||
parsed.owner, parsed.repo, parsed.pull_number, page=parsed.page, limit=parsed.limit
|
||||
)
|
||||
normalized = [
|
||||
{
|
||||
"filename": item.get("filename", ""),
|
||||
"status": item.get("status", ""),
|
||||
"additions": item.get("additions", 0),
|
||||
"deletions": item.get("deletions", 0),
|
||||
"changes": item.get("changes", 0),
|
||||
}
|
||||
for item in files
|
||||
if isinstance(item, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"pull_number": parsed.pull_number,
|
||||
"files": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list pull request files: {exc}") from exc
|
||||
|
||||
|
||||
async def list_pull_request_commits_tool(
|
||||
gitea: GiteaClient, arguments: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""List commits in a pull request."""
|
||||
parsed = ListPullRequestCommitsArgs.model_validate(arguments)
|
||||
try:
|
||||
commits = await gitea.list_pull_request_commits(
|
||||
parsed.owner, parsed.repo, parsed.pull_number, page=parsed.page, limit=parsed.limit
|
||||
)
|
||||
normalized = [
|
||||
{
|
||||
"sha": commit.get("sha", ""),
|
||||
"message": limit_text(str(commit.get("commit", {}).get("message", ""))),
|
||||
"author": (
|
||||
commit.get("author", {}).get("login", "")
|
||||
if isinstance(commit.get("author"), dict)
|
||||
else ""
|
||||
),
|
||||
}
|
||||
for commit in commits
|
||||
if isinstance(commit, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"pull_number": parsed.pull_number,
|
||||
"commits": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list pull request commits: {exc}") from exc
|
||||
|
||||
|
||||
async def list_issue_comments_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List comments on an issue or pull request."""
|
||||
parsed = ListIssueCommentsArgs.model_validate(arguments)
|
||||
try:
|
||||
comments = await gitea.list_issue_comments(
|
||||
parsed.owner, parsed.repo, parsed.issue_number, page=parsed.page, limit=parsed.limit
|
||||
)
|
||||
normalized = [
|
||||
{
|
||||
"id": comment.get("id", 0),
|
||||
"author": (
|
||||
comment.get("user", {}).get("login", "")
|
||||
if isinstance(comment.get("user"), dict)
|
||||
else ""
|
||||
),
|
||||
"body": limit_text(str(comment.get("body", ""))),
|
||||
"created_at": comment.get("created_at", ""),
|
||||
"updated_at": comment.get("updated_at", ""),
|
||||
"url": comment.get("html_url", ""),
|
||||
}
|
||||
for comment in comments
|
||||
if isinstance(comment, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"issue_number": parsed.issue_number,
|
||||
"comments": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list issue comments: {exc}") from exc
|
||||
|
||||
|
||||
def _normalize_branch(branch: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Normalize a Gitea branch payload."""
|
||||
commit = branch.get("commit", {}) if isinstance(branch.get("commit"), dict) else {}
|
||||
return {
|
||||
"name": branch.get("name", ""),
|
||||
"protected": branch.get("protected", False),
|
||||
"commit": commit.get("id", ""),
|
||||
}
|
||||
|
||||
|
||||
async def list_branches_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List repository branches."""
|
||||
parsed = ListBranchesArgs.model_validate(arguments)
|
||||
try:
|
||||
branches = await gitea.list_branches(
|
||||
parsed.owner, parsed.repo, page=parsed.page, limit=parsed.limit
|
||||
)
|
||||
normalized = [_normalize_branch(b) for b in branches if isinstance(b, dict)]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"branches": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list branches: {exc}") from exc
|
||||
|
||||
|
||||
async def get_branch_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get a single branch."""
|
||||
parsed = GetBranchArgs.model_validate(arguments)
|
||||
try:
|
||||
branch = await gitea.get_branch(parsed.owner, parsed.repo, parsed.branch)
|
||||
result = _normalize_branch(branch)
|
||||
result.update({"owner": parsed.owner, "repo": parsed.repo})
|
||||
return result
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to get branch: {exc}") from exc
|
||||
|
||||
|
||||
def _normalize_release(release: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Normalize a Gitea release payload."""
|
||||
return {
|
||||
"id": release.get("id", 0),
|
||||
"tag_name": release.get("tag_name", ""),
|
||||
"name": limit_text(str(release.get("name", ""))),
|
||||
"draft": release.get("draft", False),
|
||||
"prerelease": release.get("prerelease", False),
|
||||
"body": limit_text(str(release.get("body", ""))),
|
||||
"created_at": release.get("created_at", ""),
|
||||
"published_at": release.get("published_at", ""),
|
||||
"url": release.get("html_url", ""),
|
||||
}
|
||||
|
||||
|
||||
async def get_release_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get a release by id."""
|
||||
parsed = GetReleaseArgs.model_validate(arguments)
|
||||
try:
|
||||
release = await gitea.get_release(parsed.owner, parsed.repo, parsed.release_id)
|
||||
return _normalize_release(release)
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to get release: {exc}") from exc
|
||||
|
||||
|
||||
async def get_latest_release_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get the latest published release."""
|
||||
parsed = LatestReleaseArgs.model_validate(arguments)
|
||||
try:
|
||||
release = await gitea.get_latest_release(parsed.owner, parsed.repo)
|
||||
return _normalize_release(release)
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to get latest release: {exc}") from exc
|
||||
|
||||
|
||||
async def list_milestones_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List repository milestones."""
|
||||
parsed = ListMilestonesArgs.model_validate(arguments)
|
||||
try:
|
||||
milestones = await gitea.list_milestones(
|
||||
parsed.owner, parsed.repo, state=parsed.state, page=parsed.page, limit=parsed.limit
|
||||
)
|
||||
normalized = [
|
||||
{
|
||||
"id": m.get("id", 0),
|
||||
"title": limit_text(str(m.get("title", ""))),
|
||||
"state": m.get("state", ""),
|
||||
"open_issues": m.get("open_issues", 0),
|
||||
"closed_issues": m.get("closed_issues", 0),
|
||||
"due_on": m.get("due_on", ""),
|
||||
}
|
||||
for m in milestones
|
||||
if isinstance(m, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"state": parsed.state,
|
||||
"milestones": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list milestones: {exc}") from exc
|
||||
|
||||
|
||||
async def get_commit_status_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get the combined commit status for a ref/sha."""
|
||||
parsed = CommitStatusArgs.model_validate(arguments)
|
||||
try:
|
||||
status = await gitea.get_commit_status(parsed.owner, parsed.repo, parsed.sha)
|
||||
statuses_raw = status.get("statuses", []) if isinstance(status, dict) else []
|
||||
statuses = [
|
||||
{
|
||||
"context": s.get("context", ""),
|
||||
"state": s.get("status", s.get("state", "")),
|
||||
"target_url": s.get("target_url", ""),
|
||||
}
|
||||
for s in statuses_raw
|
||||
if isinstance(s, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(statuses)
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"sha": parsed.sha,
|
||||
"state": status.get("state", "") if isinstance(status, dict) else "",
|
||||
"statuses": bounded,
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to get commit status: {exc}") from exc
|
||||
|
||||
|
||||
def _normalize_repo_summary(repo: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Normalize a repository payload to a compact summary."""
|
||||
owner = repo.get("owner", {})
|
||||
return {
|
||||
"owner": owner.get("login", "") if isinstance(owner, dict) else "",
|
||||
"name": repo.get("name", ""),
|
||||
"full_name": repo.get("full_name", ""),
|
||||
"private": repo.get("private", False),
|
||||
"description": limit_text(str(repo.get("description", ""))),
|
||||
"url": repo.get("html_url", ""),
|
||||
}
|
||||
|
||||
|
||||
async def list_org_repositories_tool(
|
||||
gitea: GiteaClient, arguments: dict[str, Any]
|
||||
) -> dict[str, Any]:
|
||||
"""List repositories belonging to an organization."""
|
||||
parsed = ListOrgRepositoriesArgs.model_validate(arguments)
|
||||
try:
|
||||
repos = await gitea.list_org_repositories(parsed.org, page=parsed.page, limit=parsed.limit)
|
||||
normalized = [_normalize_repo_summary(r) for r in repos if isinstance(r, dict)]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"org": parsed.org,
|
||||
"repositories": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list org repositories: {exc}") from exc
|
||||
|
||||
|
||||
async def list_organizations_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List organizations the authenticated user belongs to."""
|
||||
parsed = ListOrganizationsArgs.model_validate(arguments)
|
||||
try:
|
||||
orgs = await gitea.list_organizations(page=parsed.page, limit=parsed.limit)
|
||||
normalized = [
|
||||
{
|
||||
"id": org.get("id", 0),
|
||||
"name": org.get("username", org.get("name", "")),
|
||||
"full_name": org.get("full_name", ""),
|
||||
"description": limit_text(str(org.get("description", ""))),
|
||||
}
|
||||
for org in orgs
|
||||
if isinstance(org, dict)
|
||||
]
|
||||
bounded, omitted = limit_items(normalized, configured_limit=parsed.limit)
|
||||
return {
|
||||
"organizations": bounded,
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list organizations: {exc}") from exc
|
||||
|
||||
|
||||
async def get_repo_languages_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Get the language breakdown for a repository."""
|
||||
parsed = RepoLanguagesArgs.model_validate(arguments)
|
||||
try:
|
||||
languages = await gitea.get_repo_languages(parsed.owner, parsed.repo)
|
||||
cleaned = {str(name): value for name, value in languages.items() if isinstance(name, str)}
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"languages": cleaned,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to get repository languages: {exc}") from exc
|
||||
|
||||
|
||||
async def list_repo_topics_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""List the topics assigned to a repository."""
|
||||
parsed = RepoTopicsArgs.model_validate(arguments)
|
||||
try:
|
||||
topics = await gitea.list_repo_topics(parsed.owner, parsed.repo)
|
||||
bounded, omitted = limit_items([{"topic": t} for t in topics])
|
||||
return {
|
||||
"owner": parsed.owner,
|
||||
"repo": parsed.repo,
|
||||
"topics": [entry["topic"] for entry in bounded],
|
||||
"count": len(bounded),
|
||||
"omitted": omitted,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to list repository topics: {exc}") from exc
|
||||
|
||||
@@ -14,13 +14,100 @@ from aegis_gitea_mcp.response_limits import limit_text
|
||||
from aegis_gitea_mcp.tools.arguments import (
|
||||
AddLabelsArgs,
|
||||
AssignIssueArgs,
|
||||
CreateBranchArgs,
|
||||
CreateIssueArgs,
|
||||
CreateIssueCommentArgs,
|
||||
CreateLabelArgs,
|
||||
CreateMilestoneArgs,
|
||||
CreatePrCommentArgs,
|
||||
CreatePullRequestArgs,
|
||||
CreateReleaseArgs,
|
||||
EditIssueCommentArgs,
|
||||
EditReleaseArgs,
|
||||
RemoveLabelsArgs,
|
||||
UpdateIssueArgs,
|
||||
UpdateLabelArgs,
|
||||
)
|
||||
|
||||
|
||||
async def create_label_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a repository label in write mode."""
|
||||
parsed = CreateLabelArgs.model_validate(arguments)
|
||||
# Gitea expects the color with a leading '#'; normalize either form.
|
||||
color = parsed.color if parsed.color.startswith("#") else f"#{parsed.color}"
|
||||
try:
|
||||
label = await gitea.create_label(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
name=parsed.name,
|
||||
color=color,
|
||||
description=parsed.description,
|
||||
exclusive=parsed.exclusive,
|
||||
)
|
||||
return {
|
||||
"id": label.get("id", 0),
|
||||
"name": limit_text(str(label.get("name", ""))),
|
||||
"color": label.get("color", ""),
|
||||
"description": limit_text(str(label.get("description", ""))),
|
||||
"url": label.get("url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
# Let auth/authz failures surface so the server returns actionable
|
||||
# re-authorization guidance instead of a generic internal error.
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to create label: {exc}") from exc
|
||||
|
||||
|
||||
async def update_label_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Update an existing repository label (located by current name)."""
|
||||
parsed = UpdateLabelArgs.model_validate(arguments)
|
||||
color = parsed.color
|
||||
if color is not None and not color.startswith("#"):
|
||||
color = f"#{color}"
|
||||
try:
|
||||
label = await gitea.update_label(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
name=parsed.name,
|
||||
new_name=parsed.new_name,
|
||||
color=color,
|
||||
description=parsed.description,
|
||||
)
|
||||
return {
|
||||
"id": label.get("id", 0),
|
||||
"name": limit_text(str(label.get("name", ""))),
|
||||
"color": label.get("color", ""),
|
||||
"description": limit_text(str(label.get("description", ""))),
|
||||
"url": label.get("url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to update label: {exc}") from exc
|
||||
|
||||
|
||||
async def remove_labels_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Remove labels (by name) from an issue or pull request."""
|
||||
parsed = RemoveLabelsArgs.model_validate(arguments)
|
||||
try:
|
||||
result = await gitea.remove_labels(
|
||||
parsed.owner, parsed.repo, parsed.issue_number, parsed.labels
|
||||
)
|
||||
remaining: list[str] = []
|
||||
if isinstance(result, list):
|
||||
remaining = [label.get("name", "") for label in result if isinstance(label, dict)]
|
||||
return {
|
||||
"issue_number": parsed.issue_number,
|
||||
"removed": parsed.labels,
|
||||
"remaining_labels": remaining,
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to remove labels: {exc}") from exc
|
||||
|
||||
|
||||
async def create_issue_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a new issue in write mode."""
|
||||
parsed = CreateIssueArgs.model_validate(arguments)
|
||||
@@ -168,3 +255,147 @@ async def assign_issue_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> di
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to assign issue: {exc}") from exc
|
||||
|
||||
|
||||
async def create_pull_request_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Open a pull request in write mode."""
|
||||
parsed = CreatePullRequestArgs.model_validate(arguments)
|
||||
try:
|
||||
pull = await gitea.create_pull_request(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
title=parsed.title,
|
||||
head=parsed.head,
|
||||
base=parsed.base,
|
||||
body=parsed.body,
|
||||
)
|
||||
return {
|
||||
"number": pull.get("number", 0),
|
||||
"title": limit_text(str(pull.get("title", ""))),
|
||||
"state": pull.get("state", ""),
|
||||
"url": pull.get("html_url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to create pull request: {exc}") from exc
|
||||
|
||||
|
||||
async def create_release_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a release in write mode."""
|
||||
parsed = CreateReleaseArgs.model_validate(arguments)
|
||||
try:
|
||||
release = await gitea.create_release(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
tag_name=parsed.tag_name,
|
||||
name=parsed.name,
|
||||
body=parsed.body,
|
||||
draft=parsed.draft,
|
||||
prerelease=parsed.prerelease,
|
||||
target=parsed.target,
|
||||
)
|
||||
return {
|
||||
"id": release.get("id", 0),
|
||||
"tag_name": release.get("tag_name", ""),
|
||||
"name": limit_text(str(release.get("name", ""))),
|
||||
"draft": release.get("draft", False),
|
||||
"prerelease": release.get("prerelease", False),
|
||||
"url": release.get("html_url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to create release: {exc}") from exc
|
||||
|
||||
|
||||
async def edit_release_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Edit an existing release in write mode."""
|
||||
parsed = EditReleaseArgs.model_validate(arguments)
|
||||
try:
|
||||
release = await gitea.edit_release(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
parsed.release_id,
|
||||
name=parsed.name,
|
||||
body=parsed.body,
|
||||
draft=parsed.draft,
|
||||
prerelease=parsed.prerelease,
|
||||
)
|
||||
return {
|
||||
"id": release.get("id", parsed.release_id),
|
||||
"tag_name": release.get("tag_name", ""),
|
||||
"name": limit_text(str(release.get("name", ""))),
|
||||
"draft": release.get("draft", False),
|
||||
"prerelease": release.get("prerelease", False),
|
||||
"url": release.get("html_url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to edit release: {exc}") from exc
|
||||
|
||||
|
||||
async def create_branch_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a branch in write mode."""
|
||||
parsed = CreateBranchArgs.model_validate(arguments)
|
||||
try:
|
||||
branch = await gitea.create_branch(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
new_branch_name=parsed.new_branch_name,
|
||||
old_branch_name=parsed.old_branch_name,
|
||||
)
|
||||
commit = branch.get("commit", {}) if isinstance(branch, dict) else {}
|
||||
return {
|
||||
"name": branch.get("name", parsed.new_branch_name),
|
||||
"commit": commit.get("id", "") if isinstance(commit, dict) else "",
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to create branch: {exc}") from exc
|
||||
|
||||
|
||||
async def create_milestone_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Create a milestone in write mode."""
|
||||
parsed = CreateMilestoneArgs.model_validate(arguments)
|
||||
try:
|
||||
milestone = await gitea.create_milestone(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
title=parsed.title,
|
||||
description=parsed.description,
|
||||
due_on=parsed.due_on,
|
||||
)
|
||||
return {
|
||||
"id": milestone.get("id", 0),
|
||||
"title": limit_text(str(milestone.get("title", ""))),
|
||||
"state": milestone.get("state", ""),
|
||||
"url": milestone.get("html_url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to create milestone: {exc}") from exc
|
||||
|
||||
|
||||
async def edit_issue_comment_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||
"""Edit an existing issue or PR comment in write mode."""
|
||||
parsed = EditIssueCommentArgs.model_validate(arguments)
|
||||
try:
|
||||
comment = await gitea.edit_issue_comment(
|
||||
parsed.owner,
|
||||
parsed.repo,
|
||||
parsed.comment_id,
|
||||
parsed.body,
|
||||
)
|
||||
return {
|
||||
"id": comment.get("id", parsed.comment_id),
|
||||
"body": limit_text(str(comment.get("body", ""))),
|
||||
"url": comment.get("html_url", ""),
|
||||
}
|
||||
except (GiteaAuthenticationError, GiteaAuthorizationError):
|
||||
raise
|
||||
except GiteaError as exc:
|
||||
raise RuntimeError(f"Failed to edit comment: {exc}") from exc
|
||||
|
||||
@@ -118,7 +118,7 @@ async def test_public_methods_delegate_to_request_and_normalize() -> None:
|
||||
if endpoint == "/api/v1/repos/acme/demo/pulls/2":
|
||||
return {"number": 2}
|
||||
if endpoint == "/api/v1/repos/acme/demo/labels":
|
||||
return [{"name": "bug"}]
|
||||
return [{"id": 1, "name": "bug"}]
|
||||
if endpoint == "/api/v1/repos/acme/demo/tags":
|
||||
return [{"name": "v1"}]
|
||||
if endpoint == "/api/v1/repos/acme/demo/releases":
|
||||
@@ -246,6 +246,51 @@ async def test_list_user_repositories_unknown_user_returns_empty() -> None:
|
||||
assert await client.list_user_repositories("ghost") == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_label_ids_maps_names_case_insensitively() -> None:
|
||||
"""Label names are resolved to ids regardless of case."""
|
||||
client = GiteaClient(token="user-token")
|
||||
|
||||
async def fake_request(method: str, endpoint: str, **kwargs):
|
||||
return [{"id": 3, "name": "Bug"}, {"id": 9, "name": "wontfix"}]
|
||||
|
||||
client._request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign]
|
||||
ids = await client._resolve_label_ids("o", "r", ["bug", "WONTFIX"], correlation_id="c")
|
||||
assert ids == [3, 9]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_resolve_label_ids_rejects_unknown_label() -> None:
|
||||
"""An unknown label name raises a clear error instead of a silent failure."""
|
||||
client = GiteaClient(token="user-token")
|
||||
|
||||
async def fake_request(method: str, endpoint: str, **kwargs):
|
||||
return [{"id": 3, "name": "bug"}]
|
||||
|
||||
client._request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign]
|
||||
with pytest.raises(GiteaError, match="Unknown label"):
|
||||
await client._resolve_label_ids("o", "r", ["ghost"], correlation_id="c")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_add_labels_resolves_names_to_ids() -> None:
|
||||
"""add_labels translates names to ids before POSTing to Gitea."""
|
||||
client = GiteaClient(token="user-token")
|
||||
captured: dict = {}
|
||||
|
||||
async def fake_request(method: str, endpoint: str, **kwargs):
|
||||
if endpoint.endswith("/labels") and method == "GET":
|
||||
return [{"id": 42, "name": "bug"}]
|
||||
if endpoint.endswith("/issues/1/labels") and method == "POST":
|
||||
captured["body"] = kwargs.get("json_body")
|
||||
return {"labels": [{"name": "bug"}]}
|
||||
return {}
|
||||
|
||||
client._request = AsyncMock(side_effect=fake_request) # type: ignore[method-assign]
|
||||
await client.add_labels("o", "r", 1, ["bug"])
|
||||
assert captured["body"] == {"labels": [42]}
|
||||
|
||||
|
||||
def test_git_refs_allow_slash_containing_refs() -> None:
|
||||
"""Legitimate refs that contain '/' validate successfully."""
|
||||
tree = FileTreeArgs(owner="o", repo="r", ref="feature/foo")
|
||||
|
||||
@@ -6,24 +6,46 @@ from aegis_gitea_mcp.config import reset_settings
|
||||
from aegis_gitea_mcp.gitea_client import GiteaError
|
||||
from aegis_gitea_mcp.tools.read_tools import (
|
||||
compare_refs_tool,
|
||||
get_branch_tool,
|
||||
get_commit_diff_tool,
|
||||
get_commit_status_tool,
|
||||
get_issue_tool,
|
||||
get_latest_release_tool,
|
||||
get_pull_request_tool,
|
||||
get_release_tool,
|
||||
get_repo_languages_tool,
|
||||
list_branches_tool,
|
||||
list_commits_tool,
|
||||
list_issue_comments_tool,
|
||||
list_issues_tool,
|
||||
list_labels_tool,
|
||||
list_milestones_tool,
|
||||
list_org_repositories_tool,
|
||||
list_organizations_tool,
|
||||
list_pull_request_commits_tool,
|
||||
list_pull_request_files_tool,
|
||||
list_pull_requests_tool,
|
||||
list_releases_tool,
|
||||
list_repo_topics_tool,
|
||||
list_tags_tool,
|
||||
search_code_tool,
|
||||
)
|
||||
from aegis_gitea_mcp.tools.write_tools import (
|
||||
add_labels_tool,
|
||||
assign_issue_tool,
|
||||
create_branch_tool,
|
||||
create_issue_comment_tool,
|
||||
create_issue_tool,
|
||||
create_label_tool,
|
||||
create_milestone_tool,
|
||||
create_pr_comment_tool,
|
||||
create_pull_request_tool,
|
||||
create_release_tool,
|
||||
edit_issue_comment_tool,
|
||||
edit_release_tool,
|
||||
remove_labels_tool,
|
||||
update_issue_tool,
|
||||
update_label_tool,
|
||||
)
|
||||
|
||||
|
||||
@@ -79,6 +101,45 @@ class StubGitea:
|
||||
async def list_releases(self, owner, repo, *, page, limit):
|
||||
return [{"id": 1, "tag_name": "v1.0.0", "name": "release"}]
|
||||
|
||||
async def list_pull_request_files(self, owner, repo, index, *, page, limit):
|
||||
return [{"filename": "a.py", "status": "modified", "additions": 1, "deletions": 0}]
|
||||
|
||||
async def list_pull_request_commits(self, owner, repo, index, *, page, limit):
|
||||
return [{"sha": "abc", "commit": {"message": "m"}, "author": {"login": "alice"}}]
|
||||
|
||||
async def list_issue_comments(self, owner, repo, index, *, page, limit):
|
||||
return [{"id": 1, "body": "hi", "user": {"login": "alice"}}]
|
||||
|
||||
async def list_branches(self, owner, repo, *, page, limit):
|
||||
return [{"name": "main", "protected": True, "commit": {"id": "abc"}}]
|
||||
|
||||
async def get_branch(self, owner, repo, branch):
|
||||
return {"name": branch, "protected": False, "commit": {"id": "abc"}}
|
||||
|
||||
async def get_release(self, owner, repo, release_id):
|
||||
return {"id": release_id, "tag_name": "v1", "name": "rel"}
|
||||
|
||||
async def get_latest_release(self, owner, repo):
|
||||
return {"id": 1, "tag_name": "v1", "name": "rel"}
|
||||
|
||||
async def list_milestones(self, owner, repo, *, state, page, limit):
|
||||
return [{"id": 1, "title": "M", "state": state}]
|
||||
|
||||
async def get_commit_status(self, owner, repo, sha):
|
||||
return {"state": "success", "statuses": [{"context": "ci", "status": "success"}]}
|
||||
|
||||
async def list_org_repositories(self, org, *, page, limit):
|
||||
return [{"name": "r", "owner": {"login": org}, "full_name": f"{org}/r"}]
|
||||
|
||||
async def list_organizations(self, *, page, limit):
|
||||
return [{"id": 1, "username": "acme", "description": "d"}]
|
||||
|
||||
async def get_repo_languages(self, owner, repo):
|
||||
return {"Python": 100, "HTML": 5}
|
||||
|
||||
async def list_repo_topics(self, owner, repo):
|
||||
return ["python", "mcp"]
|
||||
|
||||
async def create_issue(self, owner, repo, *, title, body, labels=None, assignees=None):
|
||||
return {"number": 1, "title": title, "state": "open"}
|
||||
|
||||
@@ -97,6 +158,42 @@ class StubGitea:
|
||||
async def assign_issue(self, owner, repo, index, assignees):
|
||||
return {"assignees": [{"login": user} for user in assignees]}
|
||||
|
||||
async def create_label(self, owner, repo, *, name, color, description="", exclusive=False):
|
||||
return {"id": 5, "name": name, "color": color, "description": description, "url": "u"}
|
||||
|
||||
async def update_label(self, owner, repo, *, name, new_name=None, color=None, description=None):
|
||||
return {
|
||||
"id": 5,
|
||||
"name": new_name or name,
|
||||
"color": color or "#ffffff",
|
||||
"description": description or "",
|
||||
}
|
||||
|
||||
async def remove_labels(self, owner, repo, index, labels):
|
||||
return []
|
||||
|
||||
async def create_pull_request(self, owner, repo, *, title, head, base, body=""):
|
||||
return {"number": 7, "title": title, "state": "open"}
|
||||
|
||||
async def create_release(
|
||||
self, owner, repo, *, tag_name, name="", body="", draft=False, prerelease=False, target=None
|
||||
):
|
||||
return {"id": 3, "tag_name": tag_name, "name": name or tag_name}
|
||||
|
||||
async def edit_release(
|
||||
self, owner, repo, release_id, *, name=None, body=None, draft=None, prerelease=None
|
||||
):
|
||||
return {"id": release_id, "tag_name": "v1", "name": name or "rel"}
|
||||
|
||||
async def create_branch(self, owner, repo, *, new_branch_name, old_branch_name=None):
|
||||
return {"name": new_branch_name, "commit": {"id": "abc"}}
|
||||
|
||||
async def create_milestone(self, owner, repo, *, title, description="", due_on=None):
|
||||
return {"id": 4, "title": title, "state": "open"}
|
||||
|
||||
async def edit_issue_comment(self, owner, repo, comment_id, body):
|
||||
return {"id": comment_id, "body": body}
|
||||
|
||||
|
||||
class ErrorGitea(StubGitea):
|
||||
"""Stub that raises backend errors for failure-mode coverage."""
|
||||
@@ -124,6 +221,35 @@ class ErrorGitea(StubGitea):
|
||||
(list_labels_tool, {"owner": "acme", "repo": "app"}, "labels"),
|
||||
(list_tags_tool, {"owner": "acme", "repo": "app"}, "tags"),
|
||||
(list_releases_tool, {"owner": "acme", "repo": "app"}, "releases"),
|
||||
(
|
||||
list_pull_request_files_tool,
|
||||
{"owner": "acme", "repo": "app", "pull_number": 1},
|
||||
"files",
|
||||
),
|
||||
(
|
||||
list_pull_request_commits_tool,
|
||||
{"owner": "acme", "repo": "app", "pull_number": 1},
|
||||
"commits",
|
||||
),
|
||||
(
|
||||
list_issue_comments_tool,
|
||||
{"owner": "acme", "repo": "app", "issue_number": 1},
|
||||
"comments",
|
||||
),
|
||||
(list_branches_tool, {"owner": "acme", "repo": "app"}, "branches"),
|
||||
(get_branch_tool, {"owner": "acme", "repo": "app", "branch": "main"}, "name"),
|
||||
(get_release_tool, {"owner": "acme", "repo": "app", "release_id": 1}, "tag_name"),
|
||||
(get_latest_release_tool, {"owner": "acme", "repo": "app"}, "tag_name"),
|
||||
(list_milestones_tool, {"owner": "acme", "repo": "app"}, "milestones"),
|
||||
(
|
||||
get_commit_status_tool,
|
||||
{"owner": "acme", "repo": "app", "sha": "abc1234"},
|
||||
"state",
|
||||
),
|
||||
(list_org_repositories_tool, {"org": "acme"}, "repositories"),
|
||||
(list_organizations_tool, {}, "organizations"),
|
||||
(get_repo_languages_tool, {"owner": "acme", "repo": "app"}, "languages"),
|
||||
(list_repo_topics_tool, {"owner": "acme", "repo": "app"}, "topics"),
|
||||
],
|
||||
)
|
||||
async def test_extended_read_tools_success(tool, args, expected_key):
|
||||
@@ -169,9 +295,82 @@ async def test_extended_read_tools_failure_mode() -> None:
|
||||
{"owner": "acme", "repo": "app", "issue_number": 1, "assignees": ["alice"]},
|
||||
"assignees",
|
||||
),
|
||||
(
|
||||
create_label_tool,
|
||||
{"owner": "acme", "repo": "app", "name": "bug", "color": "#ff0000"},
|
||||
"id",
|
||||
),
|
||||
(
|
||||
update_label_tool,
|
||||
{"owner": "acme", "repo": "app", "name": "bug", "new_name": "defect"},
|
||||
"id",
|
||||
),
|
||||
(
|
||||
remove_labels_tool,
|
||||
{"owner": "acme", "repo": "app", "issue_number": 1, "labels": ["bug"]},
|
||||
"removed",
|
||||
),
|
||||
(
|
||||
create_pull_request_tool,
|
||||
{"owner": "acme", "repo": "app", "title": "PR", "head": "feature", "base": "main"},
|
||||
"number",
|
||||
),
|
||||
(
|
||||
create_release_tool,
|
||||
{"owner": "acme", "repo": "app", "tag_name": "v1.0.0"},
|
||||
"id",
|
||||
),
|
||||
(
|
||||
edit_release_tool,
|
||||
{"owner": "acme", "repo": "app", "release_id": 3, "name": "x"},
|
||||
"id",
|
||||
),
|
||||
(
|
||||
create_branch_tool,
|
||||
{"owner": "acme", "repo": "app", "new_branch_name": "feature/x"},
|
||||
"name",
|
||||
),
|
||||
(
|
||||
create_milestone_tool,
|
||||
{"owner": "acme", "repo": "app", "title": "M1"},
|
||||
"id",
|
||||
),
|
||||
(
|
||||
edit_issue_comment_tool,
|
||||
{"owner": "acme", "repo": "app", "comment_id": 5, "body": "edited"},
|
||||
"id",
|
||||
),
|
||||
],
|
||||
)
|
||||
async def test_write_tools_success(tool, args, expected_key):
|
||||
"""Write tools should normalize successful backend responses."""
|
||||
result = await tool(StubGitea(), args)
|
||||
assert expected_key in result
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_label_normalizes_color_without_hash() -> None:
|
||||
"""A hex color without a leading '#' is normalized before hitting Gitea."""
|
||||
captured: dict = {}
|
||||
|
||||
class CaptureStub(StubGitea):
|
||||
async def create_label(self, owner, repo, *, name, color, description="", exclusive=False):
|
||||
captured["color"] = color
|
||||
return {"id": 7, "name": name, "color": color}
|
||||
|
||||
result = await create_label_tool(
|
||||
CaptureStub(),
|
||||
{"owner": "acme", "repo": "app", "name": "bug", "color": "ff0000"},
|
||||
)
|
||||
assert captured["color"] == "#ff0000"
|
||||
assert result["id"] == 7
|
||||
|
||||
|
||||
def test_create_label_args_reject_invalid_color() -> None:
|
||||
"""Non-hex color values are rejected at the argument layer."""
|
||||
import pydantic
|
||||
|
||||
from aegis_gitea_mcp.tools.arguments import CreateLabelArgs
|
||||
|
||||
with pytest.raises(pydantic.ValidationError):
|
||||
CreateLabelArgs(owner="o", repo="r", name="bug", color="red")
|
||||
|
||||
Reference in New Issue
Block a user