feat: complete label management (name->id resolution, update/remove)

Resolves the long-standing problem that label tools passed names while Gitea's
API requires numeric label ids.

- gitea_client: add _resolve_label_ids() helper; create_issue and add_labels now
  resolve label names to ids (case-insensitive) and raise a clear "Unknown
  label(s)" error instead of a generic 500.
- New tools: remove_labels (by name) and update_label (located by current name).
- Register both write tools and document the name-based label contract.
- Tests: resolver mapping + unknown-label error, add_labels id translation,
  update_label and remove_labels handlers.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-06-14 20:34:35 +02:00
parent f0db219ee8
commit c282ffe359
8 changed files with 306 additions and 11 deletions
+119 -9
View File
@@ -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 {}
@@ -702,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 {}
+34
View File
@@ -392,6 +392,40 @@ AVAILABLE_TOOLS: list[MCPTool] = [
},
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,
),
]
+4
View File
@@ -85,7 +85,9 @@ from aegis_gitea_mcp.tools.write_tools import (
create_issue_tool,
create_label_tool,
create_pr_comment_tool,
remove_labels_tool,
update_issue_tool,
update_label_tool,
)
logger = logging.getLogger(__name__)
@@ -344,6 +346,8 @@ TOOL_HANDLERS: dict[str, ToolHandler] = {
"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,
}
+23
View File
@@ -230,6 +230,29 @@ class CreateLabelArgs(RepositoryArgs):
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)
def extract_repository(arguments: dict[str, object]) -> str | None:
"""Extract `owner/repo` from raw argument mapping.
+51
View File
@@ -18,7 +18,9 @@ from aegis_gitea_mcp.tools.arguments import (
CreateIssueCommentArgs,
CreateLabelArgs,
CreatePrCommentArgs,
RemoveLabelsArgs,
UpdateIssueArgs,
UpdateLabelArgs,
)
@@ -51,6 +53,55 @@ async def create_label_tool(gitea: GiteaClient, arguments: dict[str, Any]) -> di
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)