gitea-mcp-remote/src/gitea_mcp/server.py

"""MCP server implementation for Gitea API integration."""

import asyncio
import argparse
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

from . import __version__
from .auth import AuthConfig
from .client import GiteaClient, GiteaClientError
from .tools import (
    get_issue_tools,
    handle_issue_tool,
    get_label_tools,
    handle_label_tool,
    get_milestone_tools,
    handle_milestone_tool,
)


# Global client instance
gitea_client: GiteaClient | None = None


async def serve() -> None:
    """Run the MCP server."""
    server = Server("gitea-mcp")

    # Initialize authentication config
    try:
        config = AuthConfig()
    except ValueError as e:
        print(f"Configuration error: {e}")
        raise

    # Initialize Gitea client
    global gitea_client
    gitea_client = GiteaClient(config)

    @server.list_tools()
    async def list_tools() -> list[Tool]:
        """List available MCP tools.

        Returns:
            list: Available tools including issue, label, and milestone operations.
        """
        # Get issue, label, and milestone tools
        tools = get_issue_tools()
        tools.extend(get_label_tools())
        tools.extend(get_milestone_tools())

        # Placeholder for future tools (PR tools, etc.)
        tools.extend([
            Tool(
                name="list_repositories",
                description="List repositories in an organization (coming soon)",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "org": {
                            "type": "string",
                            "description": "Organization name",
                        }
                    },
                    "required": ["org"],
                },
            ),
            Tool(
                name="create_pull_request",
                description="Create a new pull request (coming soon)",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "owner": {
                            "type": "string",
                            "description": "Repository owner",
                        },
                        "repo": {
                            "type": "string",
                            "description": "Repository name",
                        },
                        "title": {
                            "type": "string",
                            "description": "Pull request title",
                        },
                        "head": {
                            "type": "string",
                            "description": "Source branch",
                        },
                        "base": {
                            "type": "string",
                            "description": "Target branch",
                        },
                    },
                    "required": ["owner", "repo", "title", "head", "base"],
                },
            ),
        ])

        return tools

    @server.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[TextContent]:
        """Handle tool calls.

        Args:
            name: Tool name.
            arguments: Tool arguments.

        Returns:
            list: Tool response.
        """
        # Handle issue tools
        if name.startswith("gitea_") and any(
            name.endswith(suffix) for suffix in ["_issues", "_issue"]
        ):
            return await handle_issue_tool(name, arguments, gitea_client)

        # Handle label tools
        if name.startswith("gitea_") and any(
            name.endswith(suffix) for suffix in ["_labels", "_label"]
        ):
            return await handle_label_tool(gitea_client, name, arguments)

        # Handle milestone tools
        if name.startswith("gitea_") and any(
            name.endswith(suffix) for suffix in ["_milestones", "_milestone"]
        ):
            return await handle_milestone_tool(name, arguments, gitea_client)

        # Placeholder for other tools
        return [
            TextContent(
                type="text",
                text=f"Tool '{name}' is not yet implemented.",
            )
        ]

    # Run the server using stdio transport
    async with stdio_server() as (read_stream, write_stream):
        await server.run(
            read_stream,
            write_stream,
            server.create_initialization_options(),
        )


def main() -> None:
    """Main entry point with CLI argument parsing."""
    parser = argparse.ArgumentParser(
        description="Gitea MCP Server - MCP server for Gitea API integration"
    )
    parser.add_argument(
        "--version",
        action="version",
        version=f"gitea-mcp {__version__}",
    )

    args = parser.parse_args()

    # Run the server
    try:
        asyncio.run(serve())
    except KeyboardInterrupt:
        print("\nServer stopped by user")
    except Exception as e:
        print(f"Server error: {e}")
        raise


if __name__ == "__main__":
    main()