"""Shared, transport-agnostic tool registry. This module is the single source of truth that maps each MCP tool name to its async handler. Both transport adapters consume it: * the HTTP/OAuth server (``server.py``), and * the local stdio adapter (``stdio_app.py``). Tool *definitions* (name, description, JSON schema, read/write flag) live in ``mcp_protocol.AVAILABLE_TOOLS``; this module binds those names to callables and exposes lookup helpers so neither adapter duplicates the tool list. It imports only core modules and never the web stack, keeping the core importable without FastAPI installed. """ from __future__ import annotations from collections.abc import Awaitable, Callable from typing import Any from aegis_gitea_mcp.gitea_client import GiteaClient from aegis_gitea_mcp.mcp_protocol import ( AVAILABLE_TOOLS, MCPTool, get_tool_by_name, ) from aegis_gitea_mcp.tools.raw_tools import raw_api_request_tool 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.repository import ( get_file_contents_tool, get_file_tree_tool, get_repository_info_tool, list_repositories_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, ) ToolHandler = Callable[[GiteaClient, dict[str, Any]], Awaitable[dict[str, Any]]] TOOL_HANDLERS: dict[str, ToolHandler] = { # Baseline read tools "list_repositories": list_repositories_tool, "get_repository_info": get_repository_info_tool, "get_file_tree": get_file_tree_tool, "get_file_contents": get_file_contents_tool, # Expanded read tools "search_code": search_code_tool, "list_commits": list_commits_tool, "get_commit_diff": get_commit_diff_tool, "compare_refs": compare_refs_tool, "list_issues": list_issues_tool, "get_issue": get_issue_tool, "list_pull_requests": list_pull_requests_tool, "get_pull_request": get_pull_request_tool, "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, "create_issue_comment": create_issue_comment_tool, "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, # Generic raw API dispatch (escape hatch). Registered as a read tool so GETs # work without write-mode; the handler authorizes writes per-method itself. "gitea_request": raw_api_request_tool, } def get_tool_handler(tool_name: str) -> ToolHandler | None: """Return the async handler bound to a tool name, or None if unknown.""" return TOOL_HANDLERS.get(tool_name) def list_tool_definitions() -> list[MCPTool]: """Return all registered tool definitions (name, schema, read/write flag).""" return list(AVAILABLE_TOOLS) __all__ = [ "AVAILABLE_TOOLS", "MCPTool", "ToolHandler", "TOOL_HANDLERS", "get_tool_by_name", "get_tool_handler", "list_tool_definitions", ]