Remove mcp-gateway/forgejo-mcp/server_wrapper.py

This commit is contained in:
Zac Gaetano 2026-03-31 15:33:08 -04:00
parent 2261ac2127
commit 06a9797f4e

View file

@ -1,106 +0,0 @@
"""
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())