diff --git a/src/aegis_gitea_mcp/server.py b/src/aegis_gitea_mcp/server.py index 1ae99c8..216b872 100644 --- a/src/aegis_gitea_mcp/server.py +++ b/src/aegis_gitea_mcp/server.py @@ -57,7 +57,13 @@ TOOL_HANDLERS = { # Authentication middleware @app.middleware("http") async def authenticate_request(request: Request, call_next): - """Authenticate all requests except health checks and root.""" + """Authenticate all requests except health checks and root. + + Supports Mixed authentication mode where: + - /mcp/tools (list tools) is publicly accessible (No Auth) + - /mcp/tool/call (execute tools) requires authentication + - /mcp/sse requires authentication + """ # Skip authentication for health check and root endpoints if request.url.path in ["/", "/health"]: return await call_next(request) @@ -66,6 +72,10 @@ async def authenticate_request(request: Request, call_next): if not request.url.path.startswith("/mcp/"): return await call_next(request) + # Mixed mode: allow /mcp/tools without authentication (for ChatGPT discovery) + if request.url.path == "/mcp/tools": + return await call_next(request) + # Extract client information client_ip = request.client.host if request.client else "unknown" user_agent = request.headers.get("user-agent", "unknown") @@ -75,7 +85,7 @@ async def authenticate_request(request: Request, call_next): api_key = auth_validator.extract_bearer_token(auth_header) # Fallback: allow API key via query parameter only for MCP endpoints - if not api_key and request.url.path in {"/mcp/tools", "/mcp/tool/call", "/mcp/sse"}: + if not api_key and request.url.path in {"/mcp/tool/call", "/mcp/sse"}: api_key = request.query_params.get("api_key") # Validate API key @@ -105,14 +115,14 @@ async def startup_event() -> None: logger.info(f"Starting AegisGitea MCP Server on {settings.mcp_host}:{settings.mcp_port}") logger.info(f"Connected to Gitea instance: {settings.gitea_base_url}") logger.info(f"Audit logging enabled: {settings.audit_log_path}") - + # Log authentication status if settings.auth_enabled: key_count = len(settings.mcp_api_keys) logger.info(f"API key authentication ENABLED ({key_count} key(s) configured)") else: logger.warning("API key authentication DISABLED - server is open to all requests!") - + # Test Gitea connection try: async with GiteaClient() as gitea: @@ -249,6 +259,7 @@ async def sse_endpoint(request: Request) -> StreamingResponse: Returns: Streaming SSE response """ + async def event_stream(): """Generate SSE events.""" # Send initial connection event @@ -259,14 +270,15 @@ async def sse_endpoint(request: Request) -> StreamingResponse: while True: if await request.is_disconnected(): break - + # Heartbeat every 30 seconds yield f"data: {{'event': 'heartbeat'}}\n\n" - + # Wait for next heartbeat (in production, this would handle actual events) import asyncio + await asyncio.sleep(30) - + except Exception as e: logger.error(f"SSE stream error: {e}") @@ -286,7 +298,7 @@ def main() -> None: import uvicorn settings = get_settings() - + uvicorn.run( "aegis_gitea_mcp.server:app", host=settings.mcp_host,