"""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_http_wrapper.config import GiteaSettings, load_settings from gitea_http_wrapper.filtering import ToolFilter from gitea_http_wrapper.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_http_wrapper.server:app", host=settings.http_host, port=settings.http_port, log_level="info", ) if __name__ == "__main__": main()