Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
7 changes: 0 additions & 7 deletions examples/server_with_custom_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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()
Expand Down
29 changes: 0 additions & 29 deletions examples/simple_custom_route.py

This file was deleted.

11 changes: 11 additions & 0 deletions src/north_mcp_python_sdk/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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()
Expand Down Expand Up @@ -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()
Comment thread
JoshBragg-Cohere marked this conversation as resolved.

def _register_health_check(self) -> None:
@self.custom_route("/health", methods=["GET"])
async def health(_: Request) -> PlainTextResponse:
return PlainTextResponse("OK")


# Convenience exports
__all__ = [
Expand Down
5 changes: 0 additions & 5 deletions tests/test_custom_routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down
31 changes: 31 additions & 0 deletions tests/test_north_mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand Down Expand Up @@ -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
Loading