ceos/fastmcp
1
0
Fork 0
mirror of https://github.com/PrefectHQ/fastmcp.git synced 2026-08-24 06:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add Scalekit Provider for Enterprise Authentication (#1927)

Browse source 
Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Ravi Madabhushi <ravi.madabhushi@scalekit.com>
Co-authored-by: Ravi Madabhushi <innovativeravi@gmail.com>
Co-authored-by: saif-at-scalekit <saif.shaik@scalekit.com>
This commit is contained in:
Akshay Parihar 2025-09-27 02:28:03 +05:30 • committed by GitHub
commit 7bcf8b562c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 674 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
docs/docs.json
View file

@ -172,6 +172,7 @@
"integrations/azure",
"integrations/descope",
"integrations/github",
"integrations/scalekit",
"integrations/google",
"integrations/workos"
]

187
docs/integrations/scalekit.mdx Normal file
View file

@ -0,0 +1,187 @@
---
title: Scalekit 🤝 FastMCP
sidebarTitle: Scalekit
description: Secure your FastMCP server with Scalekit
icon: shield-check
tag: NEW
---
import { VersionBadge } from "/snippets/version-badge.mdx"
<VersionBadge version="2.12.5" />
Install auth stack to your FastMCP server with [Scalekit](https://scalekit.com) using the [Remote OAuth](/servers/auth/remote-oauth) pattern: Scalekit handles user authentication, and the MCP server validates issued tokens.
## Configuration
### Prerequisites
Before you begin
1. Get a [Scalekit account](https://app.scalekit.com/) and grab API credentials such as **Client ID**, **Client Secret** and **Environment URL** from _Dashboard > Developers > Settings_.
2. Have your FastMCP server's endpoint ready (can be localhost for development, e.g., `http://localhost:8000/mcp`)
### Step 1: Configure MCP server in Scalekit environment
<Steps>
<Step title="Register MCP server and set environment">
In your Scalekit dashboard:
1. Open the **MCP Servers** section, then select **Create new server**
2. Enter server details: a name, a resource identifier, and the desired MCP client authentication settings
3. Save, then copy the **Resource ID** (for example, res_92015146095)
In your FastMCP project's `.env`:
```sh
SCALEKIT_ENVIRONMENT_URL=<YOUR_APP_ENVIRONMENT_URL>
SCALEKIT_CLIENT_ID=<YOUR_APP_CLIENT_ID> # skc_7008EXAMPLE46
SCALEKIT_RESOURCE_ID=<YOUR_APP_RESOURCE_ID> # res_926EXAMPLE5878
MCP_URL=http://localhost:8000/mcp
```
</Step>
</Steps>
### Step 2: Add auth to FastMCP server
Create your FastMCP server file and use the ScalekitProvider to handle all the OAuth integration automatically:
```python server.py
from fastmcp import FastMCP
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
# Discovers Scalekit endpoints and set up JWT token validation
auth_provider = ScalekitProvider(
environment_url=SCALEKIT_ENVIRONMENT_URL, # Scalekit environment URL
client_id=SCALEKIT_CLIENT_ID, # OAuth client ID
resource_id=SCALEKIT_RESOURCE_ID, # Resource server ID
mcp_url=SERVER_URL, # Is also aud claim
)
# Create FastMCP server with auth
mcp = FastMCP(name="My Scalekit Protected Server", auth=auth_provider)
@mcp.tool
def auth_status() -> dict:
"""Show Scalekit authentication status."""
# Extract user claims from the JWT
return {
"message": "This tool requires authentication via Scalekit",
"authenticated": True,
"provider": "Scalekit"
}
```
## Testing
### Start the MCP server
```sh
uv run python server.py
```
Use any MCP client (for example, mcp-inspector, Claude, VS Code, or Windsurf) to connect to the running serve. Verify that authentication succeeds and requests are authorized as expected.
### Provider selection
Setting this environment variable allows the Scalekit provider to be used automatically without explicitly instantiating it in code.
<Card>
<ParamField path="FASTMCP_SERVER_AUTH" default="Not set">
Set to `fastmcp.server.auth.providers.scalekit.ScalekitProvider` to use Scalekit authentication.
</ParamField>
</Card>
### Scalekit-specific configuration
These environment variables provide default values for the Scalekit provider, whether it's instantiated manually or configured via `FASTMCP_SERVER_AUTH`.
<Card>
<ParamField path="FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_ENVIRONMENT_URL" required>
Your Scalekit environment URL from the Admin Portal (e.g., `https://your-env.scalekit.com`)
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_CLIENT_ID" required>
Your Scalekit OAuth application client ID from the Applications section
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_RESOURCE_ID" required>
Your Scalekit resource server ID from the Resources section
</ParamField>
<ParamField path="FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_MCP_URL" required>
Public URL of your FastMCP server (e.g., `https://your-server.com` or `http://localhost:8000/mcp` for development)
</ParamField>
</Card>
Example `.env`:
```bash
# Use the Scalekit provider
FASTMCP_SERVER_AUTH=fastmcp.server.auth.providers.scalekit.ScalekitProvider
# Scalekit configuration
FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_ENVIRONMENT_URL=https://your-env.scalekit.com
FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_CLIENT_ID=skc_123
FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_RESOURCE_ID=res_456
FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_MCP_URL=https://your-server.com/mcp
```
With environment variables set, your server code simplifies to:
```python server.py
from fastmcp import FastMCP
# Authentication is automatically configured from environment
mcp = FastMCP(name="My Scalekit Protected Server")
@mcp.tool
def protected_action() -> str:
"""A tool that requires authentication."""
return "Access granted via Scalekit!"
```
## Capabilities
Scalekit supports OAuth 2.1 with Dynamic Client Registration for MCP clients and enterprise SSO, and provides built‑in JWT validation and security controls.
**OAuth 2.1/DCR**: clients self‑register, use PKCE, and work with the Remote OAuth pattern without pre‑provisioned credentials.
**Validation and SSO**: tokens are verified (keys, RS256, issuer, audience, expiry), and SAML, OIDC, OAuth 2.0, ADFS, Azure AD, and Google Workspace are supported; use HTTPS in production and review auth logs as needed.
## Debugging
Enable detailed logging to troubleshoot authentication issues:
```python
import logging
logging.basicConfig(level=logging.DEBUG)
```
### Token inspection
You can inspect JWT tokens in your tools to understand the user context:
```python
from fastmcp.server.context import request_ctx
import jwt
@mcp.tool
def inspect_token() -> dict:
"""Inspect the current JWT token claims."""
context = request_ctx.get()
# Extract token from Authorization header
if hasattr(context, 'request') and hasattr(context.request, 'headers'):
auth_header = context.request.headers.get('authorization', '')
if auth_header.startswith('Bearer '):
token = auth_header[7:]
# Decode without verification (already verified by provider)
claims = jwt.decode(token, options={"verify_signature": False})
return claims
return {"error": "No token found"}
```

54
examples/auth/scalekit_oauth/README.md Normal file
View file

@ -0,0 +1,54 @@
# Scalekit OAuth Example
Demonstrates FastMCP server protection with Scalekit OAuth.
## Setup
### 1. Configure MCP server in Scalekit environment
**Create a Scalekit Account**:
- Go to [Scalekit Dashboard](https://app.scalekit.com/)
- Navigate to **Developers** → **Settings**
- Copy your Environment URL, Client ID, and Client Secret
**Register Your MCP Server**:
- Go to **MCP Servers** → **Create New Server**
- Fill in your MCP server details
- Note the **Resource ID** (e.g., `res_123`)
Create a `.env` file:
```bash
# Required Scalekit credentials
SCALEKIT_ENVIRONMENT_URL=<YOUR_APP_ENVIRONMENT_URL>
SCALEKIT_CLIENT_ID=<YOUR_APP_CLIENT_ID> # skc_7008EXAMPLE46
SCALEKIT_RESOURCE_ID=<YOUR_APP_RESOURCE_ID> # res_926EXAMPLE5878
MCP_URL=http://localhost:8000/mcp
```
### 2. Run the Example
Start the server:
```bash
# From this directory
uv run python server.py
```
The server will start on `http://localhost:8000/mcp` with Scalekit OAuth authentication enabled.
Test with client:
```bash
uv run python client.py
```
The `client.py` will:
1. Attempt to connect to the server
2. Detect that OAuth authentication is required
3. Open a browser for Scalekit authentication
4. Complete the OAuth flow and connect to the server
5. Demonstrate calling authenticated tools

41
examples/auth/scalekit_oauth/client.py Normal file
View file

@ -0,0 +1,41 @@
"""OAuth client example for connecting to Scalekit-protected FastMCP servers.
This example demonstrates how to connect to a Scalekit OAuth-protected FastMCP server.
To run:
python client.py
"""
import asyncio
from fastmcp.client import Client
SERVER_URL = "http://127.0.0.1:8000/mcp"
async def main():
try:
async with Client(SERVER_URL, auth="oauth") as client:
assert await client.ping()
print("✅ Successfully authenticated with Scalekit!")
tools = await client.list_tools()
print(f"🔧 Available tools ({len(tools)}):")
for tool in tools:
print(f" - {tool.name}: {tool.description}")
# Test calling a tool
result = await client.call_tool("echo", {"message": "Hello from Scalekit!"})
print(f"🎯 Echo result: {result}")
# Test calling auth status tool
auth_status = await client.call_tool("auth_status", {})
print(f"👤 Auth status: {auth_status}")
except Exception as e:
print(f"❌ Authentication failed: {e}")
raise
if __name__ == "__main__":
asyncio.run(main())

48
examples/auth/scalekit_oauth/server.py Normal file
View file

@ -0,0 +1,48 @@
"""Scalekit OAuth server example for FastMCP.
This example demonstrates how to protect a FastMCP server with Scalekit OAuth.
Required environment variables:
- SCALEKIT_ENVIRONMENT_URL: Your Scalekit environment URL (e.g., "https://your-env.scalekit.com")
- SCALEKIT_CLIENT_ID: Your Scalekit OAuth application client ID
- SCALEKIT_RESOURCE_ID: Your Scalekit resource ID
To run:
python server.py
"""
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
auth = ScalekitProvider(
environment_url=os.getenv("SCALEKIT_ENVIRONMENT_URL")
or "https://your-env.scalekit.com",
client_id=os.getenv("SCALEKIT_CLIENT_ID") or "",
resource_id=os.getenv("SCALEKIT_RESOURCE_ID") or "",
mcp_url=os.getenv("MCP_URL", "http://localhost:8000/mcp"),
)
mcp = FastMCP("Scalekit OAuth Example Server", auth=auth)
@mcp.tool
def echo(message: str) -> str:
"""Echo the provided message."""
return message
@mcp.tool
def auth_status() -> dict:
"""Show Scalekit authentication status."""
# In a real implementation, you would extract user info from the JWT token
return {
"message": "This tool requires authentication via Scalekit",
"authenticated": True,
"provider": "Scalekit",
}
if __name__ == "__main__":
mcp.run(transport="http", port=8000)

181
src/fastmcp/server/auth/providers/scalekit.py Normal file
View file

@ -0,0 +1,181 @@
"""Scalekit authentication provider for FastMCP.
This module provides ScalekitProvider - a complete authentication solution that integrates
with Scalekit's OAuth 2.1 and OpenID Connect services, supporting Resource Server
authentication for seamless MCP client authentication.
"""
from __future__ import annotations
from typing import Any
import httpx
from pydantic import AnyHttpUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.responses import JSONResponse
from starlette.routing import Route
from fastmcp.server.auth import RemoteAuthProvider, TokenVerifier
from fastmcp.server.auth.providers.jwt import JWTVerifier
from fastmcp.utilities.logging import get_logger
from fastmcp.utilities.types import NotSet, NotSetT
logger = get_logger(__name__)
class ScalekitProviderSettings(BaseSettings):
model_config = SettingsConfigDict(
env_prefix="FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_",
env_file=".env",
extra="ignore",
)
environment_url: AnyHttpUrl
client_id: str
resource_id: str
mcp_url: AnyHttpUrl
class ScalekitProvider(RemoteAuthProvider):
"""Scalekit resource server provider for OAuth 2.1 authentication.
This provider implements Scalekit integration using resource server pattern.
FastMCP acts as a protected resource server that validates access tokens issued
by Scalekit's authorization server.
IMPORTANT SETUP REQUIREMENTS:
1. Create an MCP Server in Scalekit Dashboard:
- Go to your [Scalekit Dashboard](https://app.scalekit.com/)
- Navigate to MCP Servers section
- Register a new MCP Server with appropriate scopes
- Ensure the Resource Identifier matches exactly what you configure as MCP URL
- Note the Resource ID
2. Environment Configuration:
- Set SCALEKIT_ENVIRONMENT_URL (e.g., https://your-env.scalekit.com)
- Set SCALEKIT_CLIENT_ID from your OAuth application
- Set SCALEKIT_RESOURCE_ID from your created resource
- Set MCP_URL to your FastMCP server's public URL
For detailed setup instructions, see:
https://docs.scalekit.com/mcp/overview/
Example:
```python
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
# Create Scalekit resource server provider
scalekit_auth = ScalekitProvider(
environment_url="https://your-env.scalekit.com",
client_id="sk_client_...",
resource_id="sk_resource_...",
mcp_url="https://your-fastmcp-server.com",
)
# Use with FastMCP
mcp = FastMCP("My App", auth=scalekit_auth)
```
"""
def __init__(
self,
*,
environment_url: AnyHttpUrl | str | NotSetT = NotSet,
client_id: str | NotSetT = NotSet,
resource_id: str | NotSetT = NotSet,
mcp_url: AnyHttpUrl | str | NotSetT = NotSet,
token_verifier: TokenVerifier | None = None,
):
"""Initialize Scalekit resource server provider.
Args:
environment_url: Your Scalekit environment URL (e.g., "https://your-env.scalekit.com")
client_id: Your Scalekit OAuth client ID
resource_id: Your Scalekit resource ID
mcp_url: Public URL of this FastMCP server (used as audience)
token_verifier: Optional token verifier. If None, creates JWT verifier for Scalekit
"""
settings = ScalekitProviderSettings.model_validate(
{
k: v
for k, v in {
"environment_url": environment_url,
"client_id": client_id,
"resource_id": resource_id,
"mcp_url": mcp_url,
}.items()
if v is not NotSet
}
)
self.environment_url = str(settings.environment_url).rstrip("/")
self.client_id = settings.client_id
self.resource_id = settings.resource_id
self.mcp_url = str(settings.mcp_url)
# Create default JWT verifier if none provided
if token_verifier is None:
token_verifier = JWTVerifier(
jwks_uri=f"{self.environment_url}/keys",
issuer=self.environment_url,
algorithm="RS256",
audience=self.mcp_url,
)
# Initialize RemoteAuthProvider with Scalekit as the authorization server
super().__init__(
token_verifier=token_verifier,
authorization_servers=[
AnyHttpUrl(f"{self.environment_url}/resources/{self.resource_id}")
],
base_url=self.mcp_url,
)
def get_routes(
self,
mcp_path: str | None = None,
mcp_endpoint: Any | None = None,
) -> list[Route]:
"""Get OAuth routes including Scalekit authorization server metadata forwarding.
This returns the standard protected resource routes plus an authorization server
metadata endpoint that forwards Scalekit's OAuth metadata to clients.
Args:
mcp_path: The path where the MCP endpoint is mounted (e.g., "/mcp")
mcp_endpoint: The MCP endpoint handler to protect with auth
"""
# Get the standard protected resource routes from RemoteAuthProvider
routes = super().get_routes(mcp_path, mcp_endpoint)
async def oauth_authorization_server_metadata(request):
"""Forward Scalekit OAuth authorization server metadata with FastMCP customizations."""
try:
async with httpx.AsyncClient() as client:
response = await client.get(
f"{self.environment_url}/.well-known/oauth-authorization-server/resources/{self.resource_id}"
)
response.raise_for_status()
metadata = response.json()
return JSONResponse(metadata)
except Exception as e:
logger.error(f"Failed to fetch Scalekit metadata: {e}")
return JSONResponse(
{
"error": "server_error",
"error_description": f"Failed to fetch Scalekit metadata: {e}",
},
status_code=500,
)
# Add Scalekit authorization server metadata forwarding
routes.append(
Route(
"/.well-known/oauth-authorization-server",
endpoint=oauth_authorization_server_metadata,
methods=["GET"],
)
)
return routes

162
tests/server/auth/providers/test_scalekit.py Normal file
View file

@ -0,0 +1,162 @@
"""Tests for Scalekit OAuth provider."""
import os
from collections.abc import Generator
from unittest.mock import patch
import httpx
import pytest
from fastmcp import Client, FastMCP
from fastmcp.client.transports import StreamableHttpTransport
from fastmcp.server.auth.providers.scalekit import ScalekitProvider
from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
class TestScalekitProvider:
"""Test Scalekit OAuth provider functionality."""
def test_init_with_explicit_params(self):
"""Test ScalekitProvider initialization with explicit parameters."""
provider = ScalekitProvider(
environment_url="https://my-env.scalekit.com",
client_id="sk_client_123",
resource_id="sk_resource_456",
mcp_url="https://myserver.com/",
)
assert provider.environment_url == "https://my-env.scalekit.com"
assert provider.client_id == "sk_client_123"
assert provider.resource_id == "sk_resource_456"
assert str(provider.mcp_url) == "https://myserver.com/"
def test_init_with_env_vars(self):
"""Test ScalekitProvider initialization from environment variables."""
with patch.dict(
os.environ,
{
"FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_ENVIRONMENT_URL": "https://env-scalekit.com",
"FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_CLIENT_ID": "skc_123",
"FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_RESOURCE_ID": "res_456",
"FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_MCP_URL": "https://envserver.com/mcp",
},
):
provider = ScalekitProvider()
assert provider.environment_url == "https://env-scalekit.com"
assert provider.client_id == "skc_123"
assert provider.resource_id == "res_456"
assert str(provider.mcp_url) == "https://envserver.com/mcp"
def test_environment_variable_loading(self):
"""Test that environment variables are loaded correctly."""
provider = ScalekitProvider(
environment_url="https://test-env.scalekit.com",
client_id="sk_client_test_123",
resource_id="sk_resource_test_456",
mcp_url="http://test-server.com",
)
assert provider.environment_url == "https://test-env.scalekit.com"
assert provider.client_id == "sk_client_test_123"
assert provider.resource_id == "sk_resource_test_456"
assert str(provider.mcp_url) == "http://test-server.com/"
def test_url_trailing_slash_handling(self):
"""Test that URLs handle trailing slashes correctly."""
provider = ScalekitProvider(
environment_url="https://my-env.scalekit.com/",
client_id="sk_client_123",
resource_id="sk_resource_456",
mcp_url="https://myserver.com/",
)
assert provider.environment_url == "https://my-env.scalekit.com"
assert str(provider.mcp_url) == "https://myserver.com/"
def test_jwt_verifier_configured_correctly(self):
"""Test that JWT verifier is configured correctly."""
provider = ScalekitProvider(
environment_url="https://my-env.scalekit.com",
client_id="sk_client_123",
resource_id="sk_resource_456",
mcp_url="https://myserver.com/",
)
# Check that JWT verifier uses the correct endpoints
assert (
provider.token_verifier.jwks_uri # type: ignore[attr-defined]
== "https://my-env.scalekit.com/keys"
)
assert (
provider.token_verifier.issuer == "https://my-env.scalekit.com" # type: ignore[attr-defined]
)
assert provider.token_verifier.audience == "https://myserver.com/" # type: ignore[attr-defined]
def test_authorization_servers_configuration(self):
"""Test that authorization servers are configured correctly."""
provider = ScalekitProvider(
environment_url="https://my-env.scalekit.com",
client_id="sk_client_123",
resource_id="sk_resource_456",
mcp_url="https://myserver.com/",
)
assert len(provider.authorization_servers) == 1
assert (
str(provider.authorization_servers[0])
== "https://my-env.scalekit.com/resources/sk_resource_456"
)
def run_mcp_server(host: str, port: int) -> None:
mcp = FastMCP(
auth=ScalekitProvider(
environment_url="https://test-env.scalekit.com",
client_id="sk_client_test_123",
resource_id="sk_resource_test_456",
mcp_url="http://localhost:4321",
)
)
@mcp.tool
def add(a: int, b: int) -> int:
return a + b
mcp.run(host=host, port=port, transport="http")
@pytest.fixture
def mcp_server_url() -> Generator[str]:
with run_server_in_process(run_mcp_server) as url:
yield f"{url}/mcp"
@pytest.fixture()
def client_with_headless_oauth(
mcp_server_url: str,
) -> Generator[Client, None, None]:
"""Client with headless OAuth that bypasses browser interaction."""
client = Client(
transport=StreamableHttpTransport(mcp_server_url),
auth=HeadlessOAuth(mcp_url=mcp_server_url),
)
yield client
class TestScalekitProviderIntegration:
async def test_unauthorized_access(self, mcp_server_url: str):
with pytest.raises(httpx.HTTPStatusError) as exc_info:
async with Client(mcp_server_url) as client:
tools = await client.list_tools() # noqa: F841
assert isinstance(exc_info.value, httpx.HTTPStatusError)
assert exc_info.value.response.status_code == 401
assert "tools" not in locals()
# async def test_authorized_access(self, client_with_headless_oauth: Client):
# async with client_with_headless_oauth:
# tools = await client_with_headless_oauth.list_tools()
# assert tools is not None
# assert len(tools) > 0
# assert "add" in tools