Add forgejo-mcp/server_wrapper.py

forgejo-mcp/server_wrapper.py

@@ -0,0 +1,106 @@
+"""
+Forgejo MCP Server Wrapper
+Runs the MCP server with HTTP transport on port 8400
+Compatible with MCP Gateway tool discovery
+"""
+
+import asyncio
+import json
+import logging
+import os
+
+import uvicorn
+from starlette.applications import Starlette
+from starlette.responses import JSONResponse
+from starlette.routing import Route
+
+from forgejo_mcp import server
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("forgejo-mcp")
+
+PORT = int(os.environ.get("PORT", "8400"))
+
+
+async def handle_list_tools(request):
+    """GET /tools - List available tools."""
+    try:
+        tools = await server.list_tools()
+        return JSONResponse(tools)
+    except Exception as e:
+        logger.error(f"Error listing tools: {e}")
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+async def handle_call_tool(request):
+    """POST /call_tool - Call a specific tool."""
+    try:
+        data = await request.json()
+        tool_name = data.get("name") or data.get("tool")
+        arguments = data.get("arguments", {})
+
+        if not tool_name:
+            return JSONResponse({"error": "Missing 'name' or 'tool' parameter"}, status_code=400)
+
+        logger.info(f"Calling tool: {tool_name} with args: {arguments}")
+        result = await server.call_tool(tool_name, arguments)
+
+        return JSONResponse({
+            "content": [{"type": "text", "text": r.text} for r in result]
+        })
+    except Exception as e:
+        logger.error(f"Tool call error: {e}", exc_info=True)
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+async def handle_health(request):
+    """GET /health - Health check endpoint."""
+    return JSONResponse({"status": "healthy"})
+
+
+async def handle_mcp_sse(request):
+    """POST /mcp - Handle MCP SSE protocol requests from gateway."""
+    try:
+        logger.info(f"MCP request: {request.method} {request.url.path}")
+
+        # For POST requests, parse the JSON body
+        if request.method == "POST":
+            try:
+                data = await request.json()
+                logger.info(f"MCP POST data: {data}")
+
+                # Check if this is a tool call request
+                if "name" in data or "tool" in data:
+                    return await handle_call_tool(request)
+            except Exception as e:
+                logger.warning(f"Could not parse JSON: {e}")
+
+        # For GET requests or other cases, list tools
+        return await handle_list_tools(request)
+
+    except Exception as e:
+        logger.error(f"MCP handler error: {e}", exc_info=True)
+        return JSONResponse({"error": str(e)}, status_code=500)
+
+
+# Create Starlette app with routes
+app = Starlette(
+    routes=[
+        Route("/tools", handle_list_tools, methods=["GET"]),
+        Route("/call_tool", handle_call_tool, methods=["POST"]),
+        Route("/health", handle_health, methods=["GET"]),
+        Route("/mcp", handle_mcp_sse, methods=["GET", "POST"]),
+    ],
+)
+
+
+if __name__ == "__main__":
+    logger.info(f"Starting Forgejo MCP server on port {PORT}")
+    config = uvicorn.Config(
+        app=app,
+        host="0.0.0.0",
+        port=PORT,
+        log_level="info",
+    )
+    server_instance = uvicorn.Server(config)
+    asyncio.run(server_instance.serve())