diff --git a/src/gitea_http_wrapper/__init__.py b/src/gitea_http_wrapper/__init__.py index d6116ea..334b12c 100644 --- a/src/gitea_http_wrapper/__init__.py +++ b/src/gitea_http_wrapper/__init__.py @@ -11,5 +11,7 @@ Architecture: - server.py: Main HTTP MCP server implementation """ +from .server import GiteaMCPWrapper, create_app, main + __version__ = "0.1.0" -__all__ = ["__version__"] +__all__ = ["__version__", "GiteaMCPWrapper", "create_app", "main"] diff --git a/src/gitea_http_wrapper/server.py b/src/gitea_http_wrapper/server.py new file mode 100644 index 0000000..8155f0b --- /dev/null +++ b/src/gitea_http_wrapper/server.py @@ -0,0 +1,309 @@ +"""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()