.
This commit is contained in:
15
src/aegis_gitea_mcp/tools/__init__.py
Normal file
15
src/aegis_gitea_mcp/tools/__init__.py
Normal file
@@ -0,0 +1,15 @@
|
||||
"""MCP tool implementations for AegisGitea."""
|
||||
|
||||
from aegis_gitea_mcp.tools.repository import (
|
||||
get_file_contents_tool,
|
||||
get_file_tree_tool,
|
||||
get_repository_info_tool,
|
||||
list_repositories_tool,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"list_repositories_tool",
|
||||
"get_repository_info_tool",
|
||||
"get_file_tree_tool",
|
||||
"get_file_contents_tool",
|
||||
]
|
||||
189
src/aegis_gitea_mcp/tools/repository.py
Normal file
189
src/aegis_gitea_mcp/tools/repository.py
Normal file
@@ -0,0 +1,189 @@
|
||||
"""Repository-related MCP tool implementations."""
|
||||
|
||||
import base64
|
||||
from typing import Any, Dict
|
||||
|
||||
from aegis_gitea_mcp.gitea_client import GiteaClient, GiteaError
|
||||
|
||||
|
||||
async def list_repositories_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""List all repositories visible to the bot user.
|
||||
|
||||
Args:
|
||||
gitea: Initialized Gitea client
|
||||
arguments: Tool arguments (empty for this tool)
|
||||
|
||||
Returns:
|
||||
Dict containing list of repositories with metadata
|
||||
"""
|
||||
try:
|
||||
repos = await gitea.list_repositories()
|
||||
|
||||
# Transform to simplified format
|
||||
simplified_repos = [
|
||||
{
|
||||
"owner": repo.get("owner", {}).get("login", ""),
|
||||
"name": repo.get("name", ""),
|
||||
"full_name": repo.get("full_name", ""),
|
||||
"description": repo.get("description", ""),
|
||||
"private": repo.get("private", False),
|
||||
"default_branch": repo.get("default_branch", "main"),
|
||||
"language": repo.get("language", ""),
|
||||
"stars": repo.get("stars_count", 0),
|
||||
"url": repo.get("html_url", ""),
|
||||
}
|
||||
for repo in repos
|
||||
]
|
||||
|
||||
return {
|
||||
"repositories": simplified_repos,
|
||||
"count": len(simplified_repos),
|
||||
}
|
||||
|
||||
except GiteaError as e:
|
||||
raise Exception(f"Failed to list repositories: {str(e)}")
|
||||
|
||||
|
||||
async def get_repository_info_tool(
|
||||
gitea: GiteaClient, arguments: Dict[str, Any]
|
||||
) -> Dict[str, Any]:
|
||||
"""Get detailed information about a specific repository.
|
||||
|
||||
Args:
|
||||
gitea: Initialized Gitea client
|
||||
arguments: Tool arguments with 'owner' and 'repo'
|
||||
|
||||
Returns:
|
||||
Dict containing repository information
|
||||
"""
|
||||
owner = arguments.get("owner")
|
||||
repo = arguments.get("repo")
|
||||
|
||||
if not owner or not repo:
|
||||
raise ValueError("Both 'owner' and 'repo' arguments are required")
|
||||
|
||||
try:
|
||||
repo_data = await gitea.get_repository(owner, repo)
|
||||
|
||||
return {
|
||||
"owner": repo_data.get("owner", {}).get("login", ""),
|
||||
"name": repo_data.get("name", ""),
|
||||
"full_name": repo_data.get("full_name", ""),
|
||||
"description": repo_data.get("description", ""),
|
||||
"private": repo_data.get("private", False),
|
||||
"fork": repo_data.get("fork", False),
|
||||
"default_branch": repo_data.get("default_branch", "main"),
|
||||
"language": repo_data.get("language", ""),
|
||||
"stars": repo_data.get("stars_count", 0),
|
||||
"forks": repo_data.get("forks_count", 0),
|
||||
"open_issues": repo_data.get("open_issues_count", 0),
|
||||
"size": repo_data.get("size", 0),
|
||||
"created_at": repo_data.get("created_at", ""),
|
||||
"updated_at": repo_data.get("updated_at", ""),
|
||||
"url": repo_data.get("html_url", ""),
|
||||
"clone_url": repo_data.get("clone_url", ""),
|
||||
}
|
||||
|
||||
except GiteaError as e:
|
||||
raise Exception(f"Failed to get repository info: {str(e)}")
|
||||
|
||||
|
||||
async def get_file_tree_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get file tree for a repository.
|
||||
|
||||
Args:
|
||||
gitea: Initialized Gitea client
|
||||
arguments: Tool arguments with 'owner', 'repo', optional 'ref' and 'recursive'
|
||||
|
||||
Returns:
|
||||
Dict containing file tree structure
|
||||
"""
|
||||
owner = arguments.get("owner")
|
||||
repo = arguments.get("repo")
|
||||
ref = arguments.get("ref", "main")
|
||||
recursive = arguments.get("recursive", False)
|
||||
|
||||
if not owner or not repo:
|
||||
raise ValueError("Both 'owner' and 'repo' arguments are required")
|
||||
|
||||
try:
|
||||
tree_data = await gitea.get_tree(owner, repo, ref, recursive)
|
||||
|
||||
# Transform tree entries to simplified format
|
||||
tree_entries = tree_data.get("tree", [])
|
||||
simplified_tree = [
|
||||
{
|
||||
"path": entry.get("path", ""),
|
||||
"type": entry.get("type", ""), # 'blob' (file) or 'tree' (directory)
|
||||
"size": entry.get("size", 0),
|
||||
"sha": entry.get("sha", ""),
|
||||
}
|
||||
for entry in tree_entries
|
||||
]
|
||||
|
||||
return {
|
||||
"owner": owner,
|
||||
"repo": repo,
|
||||
"ref": ref,
|
||||
"tree": simplified_tree,
|
||||
"count": len(simplified_tree),
|
||||
}
|
||||
|
||||
except GiteaError as e:
|
||||
raise Exception(f"Failed to get file tree: {str(e)}")
|
||||
|
||||
|
||||
async def get_file_contents_tool(gitea: GiteaClient, arguments: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""Get contents of a file in a repository.
|
||||
|
||||
Args:
|
||||
gitea: Initialized Gitea client
|
||||
arguments: Tool arguments with 'owner', 'repo', 'filepath', optional 'ref'
|
||||
|
||||
Returns:
|
||||
Dict containing file contents and metadata
|
||||
"""
|
||||
owner = arguments.get("owner")
|
||||
repo = arguments.get("repo")
|
||||
filepath = arguments.get("filepath")
|
||||
ref = arguments.get("ref", "main")
|
||||
|
||||
if not owner or not repo or not filepath:
|
||||
raise ValueError("'owner', 'repo', and 'filepath' arguments are required")
|
||||
|
||||
try:
|
||||
file_data = await gitea.get_file_contents(owner, repo, filepath, ref)
|
||||
|
||||
# Content is base64-encoded by Gitea
|
||||
content_b64 = file_data.get("content", "")
|
||||
encoding = file_data.get("encoding", "base64")
|
||||
|
||||
# Decode if base64
|
||||
content = content_b64
|
||||
if encoding == "base64":
|
||||
try:
|
||||
content_bytes = base64.b64decode(content_b64)
|
||||
# Try to decode as UTF-8 text
|
||||
try:
|
||||
content = content_bytes.decode("utf-8")
|
||||
except UnicodeDecodeError:
|
||||
# If not text, keep as base64
|
||||
content = content_b64
|
||||
except Exception:
|
||||
# If decode fails, keep as-is
|
||||
pass
|
||||
|
||||
return {
|
||||
"owner": owner,
|
||||
"repo": repo,
|
||||
"filepath": filepath,
|
||||
"ref": ref,
|
||||
"content": content,
|
||||
"encoding": encoding,
|
||||
"size": file_data.get("size", 0),
|
||||
"sha": file_data.get("sha", ""),
|
||||
"url": file_data.get("html_url", ""),
|
||||
}
|
||||
|
||||
except GiteaError as e:
|
||||
raise Exception(f"Failed to get file contents: {str(e)}")
|
||||
Reference in New Issue
Block a user