generated from personal-projects/leo-claude-mktplace
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 <noreply@anthropic.com>
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.
|
||||
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"]
|
||||
|
||||
@@ -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