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

"""HTTP MCP server implementation wrapping Gitea MCP."""

import asyncio
import json
import logging
import os
import sys
from pathlib import Path
from typing import Any

import uvicorn
from mcp.server import Server
from mcp.server.stdio import stdio_server
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from gitea_mcp_remote.config import GiteaSettings, load_settings
from gitea_mcp_remote.filtering import ToolFilter
from gitea_mcp_remote.middleware import (
    BearerAuthMiddleware,
    HealthCheckBypassMiddleware,
)

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)


class GiteaMCPWrapper:
    """
    HTTP wrapper around the official Gitea MCP server.

    This class manages:
    1. Starting the Gitea MCP server as a subprocess with stdio transport
    2. Proxying HTTP requests to the MCP server
    3. Filtering tools based on configuration
    4. Handling responses and errors
    """

    def __init__(self, settings: GiteaSettings):
        """
        Initialize the MCP wrapper.

        Args:
            settings: Configuration settings for Gitea and HTTP server.
        """
        self.settings = settings
        self.tool_filter = ToolFilter(
            enabled_tools=settings.enabled_tools_list,
            disabled_tools=settings.disabled_tools_list,
        )
        self.process = None
        self.reader = None
        self.writer = None

    async def start_gitea_mcp(self) -> None:
        """
        Start the Gitea MCP server as a subprocess.

        The server runs with stdio transport, and we communicate via stdin/stdout.
        """
        logger.info("Starting Gitea MCP server subprocess")

        # Set environment variables for Gitea MCP
        env = os.environ.copy()
        env.update(self.settings.get_gitea_mcp_env())

        # Start the process
        # Note: This assumes gitea-mcp-server is installed and on PATH
        # In production Docker, this should be guaranteed
        try:
            self.process = await asyncio.create_subprocess_exec(
                "gitea-mcp-server",
                stdin=asyncio.subprocess.PIPE,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
                env=env,
            )
            self.reader = self.process.stdout
            self.writer = self.process.stdin
            logger.info("Gitea MCP server started successfully")
        except FileNotFoundError:
            logger.error("gitea-mcp-server not found in PATH")
            raise RuntimeError(
                "gitea-mcp-server not found. Ensure it's installed: pip install gitea-mcp-server"
            )

    async def stop_gitea_mcp(self) -> None:
        """Stop the Gitea MCP server subprocess."""
        if self.process:
            logger.info("Stopping Gitea MCP server subprocess")
            self.process.terminate()
            await self.process.wait()
            logger.info("Gitea MCP server stopped")

    async def send_mcp_request(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
        """
        Send a JSON-RPC request to the MCP server.

        Args:
            method: MCP method name (e.g., "tools/list", "tools/call").
            params: Method parameters.

        Returns:
            JSON-RPC response from MCP server.

        Raises:
            RuntimeError: If MCP server is not running or communication fails.
        """
        if not self.writer or not self.reader:
            raise RuntimeError("MCP server not started")

        # Build JSON-RPC request
        request = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": method,
            "params": params,
        }

        # Send request
        request_json = json.dumps(request) + "\n"
        self.writer.write(request_json.encode())
        await self.writer.drain()

        # Read response
        response_line = await self.reader.readline()
        response = json.loads(response_line.decode())

        # Check for JSON-RPC error
        if "error" in response:
            logger.error(f"MCP error: {response['error']}")
            raise RuntimeError(f"MCP error: {response['error']}")

        return response.get("result", {})

    async def list_tools(self) -> dict[str, Any]:
        """
        List available tools from MCP server with filtering applied.

        Returns:
            Filtered tools list response.
        """
        response = await self.send_mcp_request("tools/list", {})
        filtered_response = self.tool_filter.filter_tools_response(response)

        logger.info(
            f"Listed {len(filtered_response.get('tools', []))} tools "
            f"(filter: {self.tool_filter.get_filter_stats()['mode']})"
        )
        return filtered_response

    async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
        """
        Call a tool on the MCP server.

        Args:
            tool_name: Name of tool to call.
            arguments: Tool arguments.

        Returns:
            Tool execution result.

        Raises:
            ValueError: If tool is filtered out.
        """
        # Check if tool is allowed
        if not self.tool_filter.should_include_tool(tool_name):
            raise ValueError(f"Tool '{tool_name}' is not available (filtered)")

        logger.info(f"Calling tool: {tool_name}")
        result = await self.send_mcp_request(
            "tools/call",
            {"name": tool_name, "arguments": arguments},
        )
        return result


# Global wrapper instance
wrapper: GiteaMCPWrapper | None = None


async def health_check(request: Request) -> JSONResponse:
    """Health check endpoint."""
    return JSONResponse({"status": "healthy"})


async def list_tools_endpoint(request: Request) -> JSONResponse:
    """List available tools."""
    try:
        tools = await wrapper.list_tools()
        return JSONResponse(tools)
    except Exception as e:
        logger.exception("Error listing tools")
        return JSONResponse(
            {"error": str(e)},
            status_code=500,
        )


async def call_tool_endpoint(request: Request) -> JSONResponse:
    """Call a tool."""
    try:
        body = await request.json()
        tool_name = body.get("name")
        arguments = body.get("arguments", {})

        if not tool_name:
            return JSONResponse(
                {"error": "Missing 'name' field"},
                status_code=400,
            )

        result = await wrapper.call_tool(tool_name, arguments)
        return JSONResponse(result)
    except ValueError as e:
        # Tool filtered
        return JSONResponse(
            {"error": str(e)},
            status_code=403,
        )
    except Exception as e:
        logger.exception("Error calling tool")
        return JSONResponse(
            {"error": str(e)},
            status_code=500,
        )


async def startup() -> None:
    """Application startup handler."""
    global wrapper
    settings = load_settings()
    wrapper = GiteaMCPWrapper(settings)
    await wrapper.start_gitea_mcp()
    logger.info(f"HTTP MCP server starting on {settings.http_host}:{settings.http_port}")


async def shutdown() -> None:
    """Application shutdown handler."""
    global wrapper
    if wrapper:
        await wrapper.stop_gitea_mcp()


# Define routes
routes = [
    Route("/health", health_check, methods=["GET"]),
    Route("/healthz", health_check, methods=["GET"]),
    Route("/ping", health_check, methods=["GET"]),
    Route("/tools/list", list_tools_endpoint, methods=["POST"]),
    Route("/tools/call", call_tool_endpoint, methods=["POST"]),
]

# Create Starlette app
app = Starlette(
    routes=routes,
    on_startup=[startup],
    on_shutdown=[shutdown],
)


def create_app(settings: GiteaSettings | None = None) -> Starlette:
    """
    Create and configure the Starlette application.

    Args:
        settings: Optional settings override for testing.

    Returns:
        Configured Starlette application.
    """
    if settings is None:
        settings = load_settings()

    # Add middleware
    app.add_middleware(HealthCheckBypassMiddleware)
    app.add_middleware(BearerAuthMiddleware, auth_token=settings.auth_token)

    return app


def main() -> None:
    """Main entry point for the HTTP MCP server."""
    settings = load_settings()

    # Log filter configuration
    filter_stats = ToolFilter(
        enabled_tools=settings.enabled_tools_list,
        disabled_tools=settings.disabled_tools_list,
    ).get_filter_stats()
    logger.info(f"Tool filtering: {filter_stats}")

    # Run server
    uvicorn.run(
        "gitea_mcp_remote.server:app",
        host=settings.http_host,
        port=settings.http_port,
        log_level="info",
    )


if __name__ == "__main__":
    main()