Freature/health

mssql_mcp_server/config/settings.py

@@ -169,7 +169,7 @@ def _load_server_config(self) -> ServerConfig:
             batch_rows_size=int(os.getenv("BATCH_ROWS_SIZE", "100")),
             enable_async=os.getenv("ENABLE_ASYNC", "true").lower() == "true",
             enable_dynamic_resources=os.getenv("ENABLE_DYNAMIC_RESOURCES", "true").lower() == "true",
-            mcp_port=int(os.getenv("FASTMCP_PORT", "8000"))
+            mcp_port=int(os.getenv("FASTMCP_PORT", "8000")),
         )
 
 

mssql_mcp_server/server.py

@@ -4,6 +4,8 @@
 from dotenv import load_dotenv
 from fastmcp import FastMCP
 from fastmcp import Context
+from starlette.requests import Request
+from starlette.responses import JSONResponse
 from mssql_mcp_server.config.settings import settings
 from mssql_mcp_server.database.async_connection import get_pool, close_pool
 from mssql_mcp_server.handlers.async_resources import AsyncResourceHandlers
@@ -118,6 +120,44 @@ async def get_object_schema_func():
         return 0
 
 
+@app.custom_route("/health", methods=["GET"])
+async def health_check(request: Request) -> JSONResponse:
+    """
+    Health check endpoint for the FastAPI application.
+
+    Args:
+        timeout: 睡眠时间（秒），每5秒打印一次日志。如果为空则直接返回结果
+
+    Returns:
+        Health status response
+    """
+
+    timeout = request.query_params.get("timeout")
+    start_time = asyncio.get_event_loop().time()
+    if timeout is None:
+        logger.info("健康检查请求 - 立即返回")
+        return JSONResponse({"status": "ok"})
+    logger.info(f"健康检查开始，超时时间: {timeout} 秒")
+    timeout = int(timeout) if timeout.isdigit() else 0
+    if timeout > 0:
+        elapsed = 0
+        while elapsed < timeout:
+            remaining = timeout - elapsed
+            wait_time = min(5, remaining)
+            await asyncio.sleep(wait_time)
+            elapsed = int(asyncio.get_event_loop().time() - start_time)
+            if elapsed < timeout:
+                logger.info(f"健康检查运行中... 已用时: {elapsed} 秒，剩余: {timeout - elapsed} 秒")
+
+    total_time = int(asyncio.get_event_loop().time() - start_time)
+    logger.info(f"健康检查完成，总耗时: {total_time} 秒")
+    return JSONResponse({
+        "status": "ok",
+        "timeout": timeout,
+        "actual_duration": total_time
+    })
+
+
 @app.tool()
 async def execute_sql(query: str, allow_modifications: bool = False, ctx: Optional[Context] = None) -> str:
     """