feat: add 13 read tools (PR files/commits, comments, branches, releases, milestones, org/status/languages/topics)
test / test (push) Successful in 1m13s
lint / lint (push) Successful in 1m14s
docker / docker-publish (pull_request) Has been skipped
docker / test (pull_request) Successful in 22s
docker / lint (pull_request) Successful in 29s
lint / lint (pull_request) Successful in 31s
test / test (pull_request) Successful in 21s
docker / docker-test (pull_request) Successful in 23s
test / test (push) Successful in 1m13s
lint / lint (push) Successful in 1m14s
docker / docker-publish (pull_request) Has been skipped
docker / test (pull_request) Successful in 22s
docker / lint (pull_request) Successful in 29s
lint / lint (pull_request) Successful in 31s
test / test (pull_request) Successful in 21s
docker / docker-test (pull_request) Successful in 23s
Expands the read surface so the MCP can inspect more of Gitea: - list_pull_request_files, list_pull_request_commits, list_issue_comments - list_branches, get_branch - get_release, get_latest_release, list_milestones - get_commit_status - list_org_repositories, list_organizations - get_repo_languages, list_repo_topics Each: arg schema (extra=forbid; GitRef on branch/sha fields), Gitea client method with url-encoded path segments, bounded handler, MCP registration (read-only), server wiring, docs, and parametrized success tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user