diff --git a/README.md b/README.md index b354a77..8ba704e 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,17 @@ This repository provides code to enable your server to use authentication with N * You can access the user's OAuth token to interact with third-party services on their behalf. * You can access the user's identity (from the identity provider used with North). * **Debug mode** for detailed authentication logging and troubleshooting. +* **Built-in health check** endpoint for Kubernetes liveness probes (enabled by default). + +## Health Check + +`NorthMCPServer` includes a built-in `/health` endpoint that responds to `GET` requests with a `200 OK` plain-text response. This is useful for Kubernetes liveness/readiness probes and load balancer health checks. The endpoint bypasses authentication, so no tokens are needed. + +It is enabled by default. To disable it: + +```python +mcp = NorthMCPServer(name="Demo", health_check=False) +``` ## Examples diff --git a/examples/server_with_custom_routes.py b/examples/server_with_custom_routes.py index a45d8fd..6ac5b82 100644 --- a/examples/server_with_custom_routes.py +++ b/examples/server_with_custom_routes.py @@ -20,12 +20,6 @@ mcp = NorthMCPServer("K8s Ready Server", port=5222) -@mcp.custom_route("/health", methods=["GET"]) -async def health_check(request: Request) -> PlainTextResponse: - """Kubernetes liveness probe - no auth required.""" - return PlainTextResponse("OK") - - @mcp.custom_route("/ready", methods=["GET"]) async def readiness_check(request: Request) -> JSONResponse: """Kubernetes readiness probe - no auth required.""" @@ -58,7 +52,6 @@ def add(a: int, b: int) -> int: print("MCP Server with Kubernetes endpoints") print() print("Public endpoints (no auth):") - print(" GET /health - Liveness probe") print(" GET /ready - Readiness probe") print(" GET /metrics - Prometheus metrics") print() diff --git a/examples/simple_custom_route.py b/examples/simple_custom_route.py deleted file mode 100644 index 281a3be..0000000 --- a/examples/simple_custom_route.py +++ /dev/null @@ -1,29 +0,0 @@ -""" -Example: Minimal Custom Route for Health Checks - -The simplest way to add a health check endpoint for Kubernetes. -Custom routes bypass authentication automatically. -""" - -from starlette.requests import Request -from starlette.responses import PlainTextResponse - -from north_mcp_python_sdk import NorthMCPServer - -mcp = NorthMCPServer("Simple Server") - - -@mcp.custom_route("/health", methods=["GET"]) -async def health(_: Request) -> PlainTextResponse: - """Health check endpoint - no authentication required.""" - return PlainTextResponse("OK") - - -@mcp.tool() -def echo(message: str) -> str: - """Echo the message back - requires authentication.""" - return f"Echo: {message}" - - -if __name__ == "__main__": - mcp.run() diff --git a/src/north_mcp_python_sdk/__init__.py b/src/north_mcp_python_sdk/__init__.py index 51a758e..b8ba99f 100644 --- a/src/north_mcp_python_sdk/__init__.py +++ b/src/north_mcp_python_sdk/__init__.py @@ -3,6 +3,8 @@ from typing import Any from fastmcp import FastMCP +from starlette.requests import Request +from starlette.responses import PlainTextResponse from .auth import NorthTokenVerifier, get_north_context @@ -23,6 +25,7 @@ def __init__( server_secret: str | None = None, trusted_issuers: list[str] | None = None, debug: bool | None = None, + health_check: bool = True, **settings: Any, ): is_debug = debug if debug is not None else is_debug_mode() @@ -56,6 +59,14 @@ def __init__( self._logger = logging.getLogger(f"NorthMCP.{name or 'Server'}") self._logger.setLevel(logging.INFO) + if health_check: + self._register_health_check() + + def _register_health_check(self) -> None: + @self.custom_route("/health", methods=["GET"]) + async def health(_: Request) -> PlainTextResponse: + return PlainTextResponse("OK") + # Convenience exports __all__ = [ diff --git a/tests/test_custom_routes.py b/tests/test_custom_routes.py index f3d3331..5dc69ee 100644 --- a/tests/test_custom_routes.py +++ b/tests/test_custom_routes.py @@ -33,11 +33,6 @@ def test_tool(message: str) -> str: else: return f"Unauthenticated tool call: {message}" - @mcp.custom_route("/health", methods=["GET"]) - async def health_check(request: Request) -> PlainTextResponse: - """Health check - should work without auth.""" - return PlainTextResponse("OK") - @mcp.custom_route("/status", methods=["GET"]) async def status_check(request: Request) -> JSONResponse: """Status check - should work without auth.""" diff --git a/tests/test_north_mcp_server.py b/tests/test_north_mcp_server.py index c1f6532..ef2f0e1 100644 --- a/tests/test_north_mcp_server.py +++ b/tests/test_north_mcp_server.py @@ -16,6 +16,11 @@ def app() -> NorthMCPServer: return NorthMCPServer() +@pytest.fixture +def app_no_health() -> NorthMCPServer: + return NorthMCPServer(health_check=False) + + @pytest_asyncio.fixture async def test_client(app: NorthMCPServer): asgi_app = app.http_app(transport="streamable-http") @@ -211,3 +216,29 @@ async def test_valid_auth_header_no_bearer( ) assert result.status_code != 401 + + +@pytest_asyncio.fixture +async def no_health_test_client(app_no_health: NorthMCPServer): + asgi_app = app_no_health.http_app(transport="streamable-http") + async with LifespanManager(asgi_app) as manager: + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=manager.app), + base_url="https://mcptest.com", + ) as client: + yield client + + +@pytest.mark.asyncio +async def test_health_check_enabled_by_default(test_client: httpx.AsyncClient): + """Built-in /health endpoint is available when health_check defaults to True.""" + result = await test_client.get("/health") + assert result.status_code == 200 + assert result.text == "OK" + + +@pytest.mark.asyncio +async def test_health_check_disabled(no_health_test_client: httpx.AsyncClient): + """/health endpoint returns 404 when health_check=False.""" + result = await no_health_test_client.get("/health") + assert result.status_code == 404