107 lines
3.2 KiB
Python
107 lines
3.2 KiB
Python
|
"""
|
||
|
|
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())
|