From fb8cc0811228116dad8d4c0b8376c963120e736f Mon Sep 17 00:00:00 2001 From: lmiranda Date: Tue, 3 Feb 2026 18:07:51 -0500 Subject: [PATCH] feat: Remove old server and create MCP base server structure - Delete old subprocess-based server.py - Create new server_http.py with base structure for MCP Streamable HTTP protocol - Update __init__.py to import from server_http - Health check endpoint in place - Middleware integration ready for MCP endpoints Issue #24 will add the actual MCP protocol endpoints. Closes #23 Co-Authored-By: Claude Opus 4.5 --- src/gitea_mcp_remote/__init__.py | 18 +- src/gitea_mcp_remote/server.py | 309 ---------------------------- src/gitea_mcp_remote/server_http.py | 69 +++++++ 3 files changed, 75 insertions(+), 321 deletions(-) delete mode 100644 src/gitea_mcp_remote/server.py create mode 100644 src/gitea_mcp_remote/server_http.py diff --git a/src/gitea_mcp_remote/__init__.py b/src/gitea_mcp_remote/__init__.py index 334b12c..7e70d08 100644 --- a/src/gitea_mcp_remote/__init__.py +++ b/src/gitea_mcp_remote/__init__.py @@ -1,17 +1,11 @@ """ -Gitea HTTP MCP Wrapper +Gitea MCP Remote — HTTP deployment wrapper for marketplace Gitea MCP server. -This package provides an HTTP transport wrapper around the official Gitea MCP server. -It handles configuration loading, tool filtering, and HTTP authentication middleware. - -Architecture: -- config/: Configuration loader module -- middleware/: HTTP authentication middleware -- filtering/: Tool filtering for Claude Desktop compatibility -- server.py: Main HTTP MCP server implementation +Imports tool definitions from gitea-mcp-server (marketplace) and serves them +over Streamable HTTP transport with authentication and TLS via Caddy. """ -from .server import GiteaMCPWrapper, create_app, main +from .server_http import create_app, main -__version__ = "0.1.0" -__all__ = ["__version__", "GiteaMCPWrapper", "create_app", "main"] +__version__ = "0.2.0" +__all__ = ["__version__", "create_app", "main"] diff --git a/src/gitea_mcp_remote/server.py b/src/gitea_mcp_remote/server.py deleted file mode 100644 index 397a5d2..0000000 --- a/src/gitea_mcp_remote/server.py +++ /dev/null @@ -1,309 +0,0 @@ -"""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() diff --git a/src/gitea_mcp_remote/server_http.py b/src/gitea_mcp_remote/server_http.py new file mode 100644 index 0000000..65e6c3d --- /dev/null +++ b/src/gitea_mcp_remote/server_http.py @@ -0,0 +1,69 @@ +""" +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 uvicorn +from starlette.applications import Starlette +from starlette.responses import JSONResponse +from starlette.routing import Route + +from gitea_mcp_remote.config import load_settings +from gitea_mcp_remote.middleware import BearerAuthMiddleware, HealthCheckBypassMiddleware +from gitea_mcp_remote.filtering import ToolFilter + +logger = logging.getLogger(__name__) + + +async def health_check(request): + """Health check endpoint - bypasses authentication.""" + return JSONResponse({"status": "ok"}) + + +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) + + # TODO: Issue #24 will add MCP protocol endpoints + # from mcp_server import get_tool_definitions, create_tool_dispatcher, GiteaClient, GiteaConfig + + routes = [ + Route("/health", health_check, methods=["GET"]), + # MCP endpoints will be added in Issue #24 + ] + + app = Starlette(routes=routes) + + # Apply middleware (order matters - health bypass first) + app = HealthCheckBypassMiddleware(app) + if settings.auth_token: + app = BearerAuthMiddleware(app, token=settings.auth_token) + + 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()