generated from personal-projects/leo-claude-mktplace
Merge feat/23: Remove old server and create MCP base server structure
This commit is contained in:
@@ -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.
|
Imports tool definitions from gitea-mcp-server (marketplace) and serves them
|
||||||
It handles configuration loading, tool filtering, and HTTP authentication middleware.
|
over Streamable HTTP transport with authentication and TLS via Caddy.
|
||||||
|
|
||||||
Architecture:
|
|
||||||
- config/: Configuration loader module
|
|
||||||
- middleware/: HTTP authentication middleware
|
|
||||||
- filtering/: Tool filtering for Claude Desktop compatibility
|
|
||||||
- server.py: Main HTTP MCP server implementation
|
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from .server import GiteaMCPWrapper, create_app, main
|
from .server_http import create_app, main
|
||||||
|
|
||||||
__version__ = "0.1.0"
|
__version__ = "0.2.0"
|
||||||
__all__ = ["__version__", "GiteaMCPWrapper", "create_app", "main"]
|
__all__ = ["__version__", "create_app", "main"]
|
||||||
|
|||||||
@@ -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()
|
|
||||||
69
src/gitea_mcp_remote/server_http.py
Normal file
69
src/gitea_mcp_remote/server_http.py
Normal file
@@ -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()
|
||||||
Reference in New Issue
Block a user