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

"""
Gitea MCP Remote — HTTP server with MCP Streamable HTTP protocol.

This module imports tool definitions from the marketplace gitea-mcp-server
package and serves them over HTTP with authentication.
"""

import logging
import asyncio
from typing import Any
import uvicorn
from starlette.applications import Starlette
from starlette.responses import JSONResponse, Response
from starlette.routing import Route, Mount
from mcp.server import Server
from mcp.server.streamable_http import StreamableHTTPServerTransport
from mcp.types import Tool, TextContent

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

# Import marketplace package
from mcp_server import get_tool_definitions, create_tool_dispatcher, GiteaClient

logger = logging.getLogger(__name__)


async def health_check(request):
    """Health check endpoint - bypasses authentication."""
    return JSONResponse({"status": "ok"})


def create_mcp_server(tool_names: list[str] | None = None) -> Server:
    """Create and configure the MCP server with tools from the marketplace."""
    mcp_server = Server("gitea-mcp-remote")

    # Get tool definitions from marketplace
    tools = get_tool_definitions(tool_filter=tool_names)

    # Create Gitea client and dispatcher
    gitea_client = GiteaClient()
    dispatcher = create_tool_dispatcher(gitea_client, tool_filter=tool_names)

    @mcp_server.list_tools()
    async def list_tools() -> list[Tool]:
        """Return available Gitea tools."""
        return tools

    @mcp_server.call_tool()
    async def call_tool(name: str, arguments: dict) -> list[TextContent]:
        """Execute a tool with the given arguments."""
        return await dispatcher(name, arguments)

    return mcp_server


def create_app():
    """Create the Starlette application with middleware."""
    settings = load_settings()

    # Set up tool filtering
    tool_filter = ToolFilter(
        enabled_tools=settings.enabled_tools_list,
        disabled_tools=settings.disabled_tools_list,
    )

    # Convert to list for marketplace API
    tool_names = None  # means "all"
    if tool_filter.enabled_tools:
        tool_names = list(tool_filter.enabled_tools)

    # Create MCP server with filtered tools
    mcp_server = create_mcp_server(tool_names)

    # Store server for endpoint access
    app_state = {"mcp_server": mcp_server}

    class MCPEndpoint:
        """ASGI app wrapper for MCP protocol endpoint."""

        async def __call__(self, scope, receive, send):
            """Handle MCP requests - both POST and HEAD."""
            method = scope.get("method", "")

            if method == "HEAD":
                # Return protocol version header
                await send({
                    "type": "http.response.start",
                    "status": 200,
                    "headers": [
                        [b"x-mcp-protocol-version", b"2024-11-05"],
                    ],
                })
                await send({
                    "type": "http.response.body",
                    "body": b"",
                })
                return

            # For POST requests, use the StreamableHTTPServerTransport
            # Create transport for this request
            transport = StreamableHTTPServerTransport(mcp_session_id=None)

            # Run the MCP server with this transport
            async with transport.connect() as (read_stream, write_stream):
                # Start server task
                server_task = asyncio.create_task(
                    app_state["mcp_server"].run(
                        read_stream,
                        write_stream,
                        app_state["mcp_server"].create_initialization_options()
                    )
                )

                # Handle the HTTP request
                try:
                    await transport.handle_request(scope, receive, send)
                finally:
                    # Cancel server task if still running
                    if not server_task.done():
                        server_task.cancel()
                        try:
                            await server_task
                        except asyncio.CancelledError:
                            pass

    # Create MCP endpoint instance
    mcp_endpoint = MCPEndpoint()

    routes = [
        Route("/health", health_check, methods=["GET"]),
        Route("/mcp", mcp_endpoint, methods=["GET", "POST", "HEAD"]),
    ]

    app = Starlette(routes=routes)

    # Apply middleware (order matters - outermost runs first)
    # HealthCheckBypass must wrap BearerAuth so it sets skip_auth flag first
    if settings.auth_token:
        app = BearerAuthMiddleware(app, auth_token=settings.auth_token)
    app = HealthCheckBypassMiddleware(app)

    return app


def main():
    """Entry point for the gitea-mcp-remote command."""
    settings = load_settings()
    app = create_app()

    logger.info(f"Starting Gitea MCP Remote on {settings.http_host}:{settings.http_port}")
    uvicorn.run(app, host=settings.http_host, port=settings.http_port)


if __name__ == "__main__":
    main()