Files
wehub-resource-sync 60e0ffc959
Upgrade checks / Notify on failure (push) Has been cancelled
Upgrade checks / Close issue on success (push) Has been cancelled
Schema Crash Test / Real-world schema crash test (232K schemas) (push) Has been cancelled
Run static analysis / static_analysis (push) Has been cancelled
Tests / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Tests / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Tests / Tests with lowest-direct dependencies (push) Has been cancelled
Tests / MCP conformance tests (push) Has been cancelled
Tests / Integration tests (push) Has been cancelled
Tests / Package install smoke (push) Has been cancelled
Upgrade checks / Static analysis (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.13 on ubuntu-latest (push) Has been cancelled
Upgrade checks / Tests: Python 3.10 on windows-latest (push) Has been cancelled
Upgrade checks / Integration tests (push) Has been cancelled
Update MCPServerConfig Schema / update-config-schema (push) Has been cancelled
Update SDK Documentation / update-sdk-docs (push) Has been cancelled
chore: import upstream snapshot with attribution
2026-07-13 12:39:59 +08:00

45 lines
1.2 KiB
Python

"""Keycloak OAuth server example for FastMCP.
This example demonstrates how to protect a FastMCP server with Keycloak OAuth.
Required: Keycloak 26.6.0 or later with Dynamic Client Registration enabled.
To run:
KEYCLOAK_REALM_URL=https://your-keycloak.com/realms/myrealm python server.py
"""
import os
from fastmcp import FastMCP
from fastmcp.server.auth.providers.keycloak import KeycloakAuthProvider
from fastmcp.server.dependencies import get_access_token
auth = KeycloakAuthProvider(
realm_url=os.getenv("KEYCLOAK_REALM_URL") or "http://localhost:8080/realms/fastmcp",
base_url="http://127.0.0.1:8000",
# audience="http://127.0.0.1:8000", # Recommended for production
)
mcp = FastMCP("Keycloak Example Server", auth=auth)
@mcp.tool
def echo(message: str) -> str:
"""Echo the provided message."""
return message
@mcp.tool
async def get_access_token_claims() -> dict:
"""Get the authenticated user's access token claims."""
token = get_access_token()
return {
"sub": token.claims.get("sub"),
"scope": token.claims.get("scope"),
"azp": token.claims.get("azp"),
}
if __name__ == "__main__":
mcp.run(transport="http", port=8000)