Documentation Index
Fetch the complete documentation index at: https://mcp-server-langgraph.mintlify.app/llms.txt
Use this file to discover all available pages before exploring further.
Keycloak Integration Guide
Complete guide to production-ready authentication with Keycloak for MCP Server with LangGraph.
Table of Contents
Overview
Keycloak integration provides production-grade authentication for your MCP server:
Key Features
- ✅ OAuth2/OIDC Compliance: Industry-standard authentication protocols
- ✅ JWKS Token Verification: No shared secrets, uses public key cryptography
- ✅ Automatic Token Refresh: Seamless session extension without re-authentication
- ✅ Role Synchronization: Automatic mapping of Keycloak roles/groups to OpenFGA
- ✅ User Provider Pattern: Pluggable architecture supporting multiple backends
- ✅ Backward Compatible: Defaults to in-memory provider for development
When to Use
| Environment | Provider | Use Case |
|---|
| Development | InMemory | Local testing, no external dependencies |
| Staging | Keycloak | Pre-production validation |
| Production | Keycloak | Enterprise authentication, SSO, compliance |
Quick Start
1. Start Keycloak
# Start infrastructure (includes Keycloak on port 8180)
make setup-infra
# Wait for Keycloak to start (60+ seconds)
# Check: http://localhost:8180
2. Initialize Keycloak
# Create realm, client, and test users
make setup-keycloak
# Output will include client secret - copy to .env
.env:
# Authentication Provider
AUTH_PROVIDER=keycloak # Switch from "inmemory" to "keycloak"
# Keycloak Configuration
KEYCLOAK_SERVER_URL=http://localhost:8180
KEYCLOAK_REALM=langgraph-agent
KEYCLOAK_CLIENT_ID=langgraph-client
KEYCLOAK_CLIENT_SECRET=<paste-from-setup-output>
KEYCLOAK_ADMIN_USERNAME=admin
KEYCLOAK_ADMIN_PASSWORD=admin
# Optional: SSL verification (disable for local dev)
KEYCLOAK_VERIFY_SSL=false
4. Test Authentication
# Run example
python examples/keycloak_usage.py
# Test users created by setup:
# - alice / alice123 (roles: user, premium)
# - bob / bob123 (roles: user)
# - admin / admin123 (roles: admin)
Architecture
Component Overview
┌─────────────────────────────────────────────────────────────┐
│ MCP Server Application │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌─────────────────┐ ┌──────────────────┐ │
│ │ AuthMiddleware │───────▶│ UserProvider │ │
│ └─────────────────┘ │ (Interface) │ │
│ │ └──────────────────┘ │
│ │ │ │
│ │ ┌────────────────┴────────────────┐ │
│ │ │ │ │
│ │ ┌──────────────┐ ┌─────────────────┐│
│ │ │ InMemory │ │ Keycloak ││
│ │ │ Provider │ │ Provider ││
│ │ └──────────────┘ └─────────────────┘│
│ │ │ │
│ │ │ │
│ ▼ ▼ │
│ ┌──────────────┐ ┌──────────────────┐ │
│ │ OpenFGA │◀──────────────────│ KeycloakClient │ │
│ │ (Authorizat.) │ │ - authenticate() │ │
│ └──────────────┘ │ - verify_token() │ │
│ │ - refresh_token()│ │
│ └──────────────────┘ │
│ │ │
└────────────────────────────────────────────────┼─────────────┘
│
▼
┌───────────────────────┐
│ Keycloak Server │
│ - OIDC/OAuth2 │
│ - User Management │
│ - JWKS Endpoints │
└───────────────────────┘
Authentication Flow (ROPC)
Client MCP Server Keycloak
│ │ │
├──POST /auth/login──────▶│ │
│ {username, password} │ │
│ ├──POST /token─────────▶│
│ │ grant_type=password │
│ │ username=alice │
│ │ password=*** │
│ │ │
│ │◀──200 OK──────────────┤
│ │ {access_token, │
│ │ refresh_token} │
│ │ │
│ ├──Verify Token────────▶│
│ │ (JWKS public key) │
│ │ │
│ ├──Sync Roles──────────▶│
│ │ (to OpenFGA) │
│ │ │
│◀──200 OK────────────────┤ │
│ {user_id, roles, │ │
│ access_token} │ │
Token Verification Flow
Client MCP Server Keycloak
│ │ │
├──Request w/ Token──────▶│ │
│ Authorization: Bearer │ │
│ │ │
│ ├──GET /certs──────────▶│
│ │ (JWKS endpoint) │
│ │ │
│ │◀──JWKS────────────────┤
│ │ {public keys} │
│ │ [cached 1 hour] │
│ │ │
│ ├──Verify Signature─────┤
│ │ (RSA public key) │
│ │ Check expiration │
│ │ Validate audience │
│ │ │
│◀──Response──────────────┤ │
Configuration
Settings Reference
| Setting | Environment Variable | Default | Description |
|---|
| Provider Type | AUTH_PROVIDER | inmemory | inmemory or keycloak |
| Server URL | KEYCLOAK_SERVER_URL | http://localhost:8180 | Keycloak base URL |
| Realm | KEYCLOAK_REALM | langgraph-agent | Keycloak realm name |
| Client ID | KEYCLOAK_CLIENT_ID | langgraph-client | OAuth2 client ID |
| Client Secret | KEYCLOAK_CLIENT_SECRET | None | OAuth2 client secret (required) |
| Admin Username | KEYCLOAK_ADMIN_USERNAME | admin | Admin API username |
| Admin Password | KEYCLOAK_ADMIN_PASSWORD | None | Admin API password |
| Verify SSL | KEYCLOAK_VERIFY_SSL | true | Verify SSL certificates |
| Timeout | KEYCLOAK_TIMEOUT | 30 | HTTP request timeout (seconds) |
Feature Flags
Control Keycloak features via environment variables with FF_ prefix:
# Keycloak features (default: all enabled)
FF_ENABLE_KEYCLOAK=true
FF_ENABLE_TOKEN_REFRESH=true
FF_KEYCLOAK_ROLE_SYNC=true
FF_OPENFGA_SYNC_ON_LOGIN=true
Programmatic Configuration
from mcp_server_langgraph.auth.keycloak import KeycloakConfig, KeycloakClient
from mcp_server_langgraph.auth.middleware import AuthMiddleware
from mcp_server_langgraph.auth.user_provider import KeycloakUserProvider
# Create Keycloak configuration
keycloak_config = KeycloakConfig(
server_url="https://keycloak.example.com",
realm="production",
client_id="mcp-server",
client_secret="your-secret-here",
admin_username="admin",
admin_password="admin-password",
verify_ssl=True,
timeout=30,
)
# Create user provider
keycloak_provider = KeycloakUserProvider(
config=keycloak_config,
openfga_client=openfga_client, # Optional
sync_on_login=True, # Sync roles to OpenFGA on login
)
# Use with AuthMiddleware
auth = AuthMiddleware(
secret_key=settings.jwt_secret_key,
openfga_client=openfga_client,
user_provider=keycloak_provider,
)
User Provider Pattern
The User Provider pattern enables pluggable authentication backends without changing application code.
Interface
All providers implement the UserProvider abstract base class:
class UserProvider(ABC):
@abstractmethod
async def authenticate(self, username: str, password: Optional[str] = None) -> Dict[str, Any]:
"""Authenticate user and return user details"""
pass
@abstractmethod
async def verify_token(self, token: str) -> Dict[str, Any]:
"""Verify JWT token"""
pass
@abstractmethod
async def get_user_by_username(self, username: str) -> Optional[Dict[str, Any]]:
"""Get user by username"""
pass
Factory Function
Use create_user_provider() to switch providers:
from mcp_server_langgraph.auth.user_provider import create_user_provider
from mcp_server_langgraph.core.config import settings
# Automatically selects provider based on settings.auth_provider
provider = create_user_provider(
provider_type=settings.auth_provider, # "inmemory" or "keycloak"
secret_key=settings.jwt_secret_key,
keycloak_config=keycloak_config if settings.auth_provider == "keycloak" else None,
openfga_client=openfga_client,
)
Switching Providers
Development → Production Migration:
# Step 1: Development with InMemory (no external dependencies)
AUTH_PROVIDER=inmemory
# Step 2: Staging with Keycloak (pre-production testing)
AUTH_PROVIDER=keycloak
KEYCLOAK_SERVER_URL=https://staging-keycloak.example.com
# Step 3: Production with Keycloak (full SSO, compliance)
AUTH_PROVIDER=keycloak
KEYCLOAK_SERVER_URL=https://keycloak.example.com
KEYCLOAK_VERIFY_SSL=true
Authentication Flows
1. Resource Owner Password Credentials (ROPC)
When to use: Direct username/password authentication
from mcp_server_langgraph.auth.keycloak import KeycloakClient
client = KeycloakClient(config)
# Authenticate user
tokens = await client.authenticate_user("alice", "alice123")
# Returns:
# {
# "access_token": "eyJhbGci...",
# "refresh_token": "eyJhbGci...",
# "expires_in": 300,
# "token_type": "Bearer"
# }
2. Token Verification (JWKS)
When to use: Verify tokens from other services
# Verify token using Keycloak public keys
payload = await client.verify_token(access_token)
# Returns decoded payload:
# {
# "sub": "550e8400-e29b-41d4-a716-446655440000",
# "preferred_username": "alice",
# "email": "alice@acme.com",
# "realm_access": {"roles": ["user", "premium"]},
# "exp": 1234567890,
# "iat": 1234567590
# }
3. Token Refresh
When to use: Extend session without re-authentication
# Refresh access token
new_tokens = await client.refresh_token(refresh_token)
# Returns new access_token and refresh_token
# Get user info from token
userinfo = await client.get_userinfo(access_token)
# Get full user details (requires admin token)
user = await client.get_user_by_username("alice")
# Returns KeycloakUser object with roles, groups, attributes
Token Management
Token Lifecycle
┌─────────────────┐
│ User Login │
│ (ROPC Flow) │
└────────┬────────┘
│
▼
┌─────────────────────────┐
│ Access Token Issued │
│ - Valid: 5 minutes │
│ - Type: JWT (RS256) │
│ - Contains: user info │
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ Token Used in Requests │
│ - Bearer Authorization │
│ - Verified via JWKS │
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ Token Expiration │
│ (5 min before exp) │
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ Refresh Token Flow │
│ - Get new access token │
│ - Extend session │
└────────┬────────────────┘
│
▼
┌─────────────────────────┐
│ Logout / Revocation │
│ - Token invalidated │
└─────────────────────────┘
JWKS Caching
To minimize HTTP calls to Keycloak:
class TokenValidator:
def __init__(self, config):
self._jwks_cache = None
self._jwks_cache_time = None
self._cache_ttl = timedelta(hours=1) # Cache public keys for 1 hour
- Reduces latency for token verification
- Decreases load on Keycloak server
- Handles key rotation automatically (refreshes if kid not found)
Token Refresh Strategy
# Automatic refresh before expiration
if token_expires_in < 300: # 5 minutes
new_tokens = await client.refresh_token(refresh_token)
# Update session with new tokens
Role Mapping to OpenFGA
Automatic Synchronization
When a user authenticates, their Keycloak roles/groups are automatically synced to OpenFGA:
async def sync_user_to_openfga(keycloak_user: KeycloakUser, openfga_client):
"""
Maps Keycloak roles/groups to OpenFGA tuples
"""
tuples = []
# Admin role → system:global admin
if "admin" in keycloak_user.realm_roles:
tuples.append({
"user": keycloak_user.user_id,
"relation": "admin",
"object": "system:global"
})
# Groups → organization memberships
for group in keycloak_user.groups:
org_name = group.strip("/").split("/")[-1]
tuples.append({
"user": keycloak_user.user_id,
"relation": "member",
"object": f"organization:{org_name}"
})
# Premium role → role assignment
if "premium" in keycloak_user.realm_roles:
tuples.append({
"user": keycloak_user.user_id,
"relation": "assignee",
"object": "role:premium"
})
await openfga_client.write_tuples(tuples)
Mapping Examples
| Keycloak | OpenFGA Tuple |
|---|
Realm role: admin | user:alice admin system:global |
Group: /acme | user:alice member organization:acme |
Group: /acme/engineering | user:alice member organization:engineering |
Realm role: premium | user:alice assignee role:premium |
Client role: executor | user:alice assignee role:executor |
Custom Mapping
To customize role mapping, modify sync_user_to_openfga():
# Example: Map custom attributes
if keycloak_user.attributes.get("department") == ["finance"]:
tuples.append({
"user": keycloak_user.user_id,
"relation": "viewer",
"object": "resource:financial-reports"
})
Troubleshooting
Common Issues
1. “Client secret required”
Problem: Keycloak authentication fails
Cause: Missing or incorrect KEYCLOAK_CLIENT_SECRET
Solution:
# Re-run setup and copy the client secret
make setup-keycloak
# Update .env
KEYCLOAK_CLIENT_SECRET=<paste-secret-here>
2. “Connection refused on port 8180”
Problem: Cannot connect to Keycloak
Cause: Keycloak not started or still initializing
Solution:
# Check if Keycloak is running
docker ps | grep keycloak
# Check Keycloak logs
docker-compose logs keycloak
# Restart if needed
docker-compose restart keycloak
# Wait 60+ seconds for initialization
3. “Token verification failed: kid not found”
Problem: Token signed with unknown key
Cause: JWKS cache out of date or key rotation
Solution:
- TokenValidator automatically refreshes JWKS
- If persistent, check Keycloak realm settings
- Verify
client_id matches token audience
4. “User not found in admin API”
Problem: User authenticated but admin API lookup fails
Cause: Admin credentials incorrect
Solution:
# Verify admin credentials in .env
KEYCLOAK_ADMIN_USERNAME=admin
KEYCLOAK_ADMIN_PASSWORD=admin
# Check Keycloak admin console
# http://localhost:8180/admin
5. “OpenFGA sync failed”
Problem: Role synchronization errors
Cause: OpenFGA not initialized or network issues
Solution:
# Verify OpenFGA is running
make setup-openfga
# Check OpenFGA connection in .env
OPENFGA_API_URL=http://localhost:8080
OPENFGA_STORE_ID=<store-id>
OPENFGA_MODEL_ID=<model-id>
# Note: Authentication succeeds even if sync fails
Debug Mode
Enable detailed logging:
# Enable debug logging
FF_ENABLE_DETAILED_LOGGING=true
LOG_LEVEL=DEBUG
# Run server
make run-streamable
Testing Connectivity
# Test Keycloak health (standard path)
curl http://localhost:8180/health/ready
# With KC_HTTP_RELATIVE_PATH=/authn (gateway routing)
curl http://localhost:8180/authn/health/ready
# Test JWKS endpoint (includes /authn prefix if configured)
curl http://localhost:8180/realms/langgraph-agent/protocol/openid-connect/certs
# Test token endpoint
curl -X POST http://localhost:8180/realms/langgraph-agent/protocol/openid-connect/token \
-d "grant_type=password" \
-d "client_id=langgraph-client" \
-d "username=alice" \
-d "password=alice123"
Production Best Practices
Security
1. Use HTTPS in Production
# Production configuration
KEYCLOAK_SERVER_URL=https://keycloak.example.com
KEYCLOAK_VERIFY_SSL=true
2. Rotate Secrets Regularly
# Rotate client secret every 90 days
# 1. Generate new secret in Keycloak admin console
# 2. Update KEYCLOAK_CLIENT_SECRET in secrets manager
# 3. Deploy updated configuration
# 4. Remove old secret after grace period
3. Use Infisical for Secret Management
from mcp_server_langgraph.secrets.manager import SecretsManager
secrets = SecretsManager()
# Store Keycloak credentials in Infisical
keycloak_config = KeycloakConfig(
server_url=secrets.get("KEYCLOAK_SERVER_URL"),
client_secret=secrets.get("KEYCLOAK_CLIENT_SECRET"),
admin_password=secrets.get("KEYCLOAK_ADMIN_PASSWORD"),
)
1. Enable JWKS Caching
- ✅ Already enabled (1-hour TTL)
- ✅ Automatic cache refresh on key rotation
2. Enable Role Sync on Login
# Sync only on login (not every request)
FF_OPENFGA_SYNC_ON_LOGIN=true
# Consider async background sync for high-traffic systems
3. Connection Pooling
# Use httpx connection pooling
client = httpx.AsyncClient(
verify=True,
timeout=30,
limits=httpx.Limits(max_connections=100, max_keepalive_connections=20)
)
Monitoring
1. Track Authentication Metrics
from mcp_server_langgraph.observability.telemetry import metrics
# Automatically tracked:
# - auth_failures (by reason: expired_token, invalid_credentials)
# - successful_calls (by operation: authenticate_user, verify_token)
# - failed_calls (by operation)
2. Set Up Alerts
- High authentication failure rate (> 10%)
- Token verification errors (> 5%)
- Keycloak connection failures
- JWKS fetch failures
3. Grafana Dashboards
See: observability/grafana/auth_dashboard.json (coming in Phase 2)
High Availability
1. Keycloak Clustering
For production, deploy Keycloak in clustered mode:
# docker-compose.prod.yml
keycloak:
# Use pre-optimized GHCR image for faster startup and readOnlyRootFilesystem support
image: ghcr.io/vishnu2kmohan/keycloak-optimized:26.4.2
command: start --optimized
environment:
- KC_DB=postgres
- KC_DB_URL_HOST=postgres-primary
- KC_CACHE=ispn
- KC_CACHE_STACK=kubernetes
- KC_HTTP_RELATIVE_PATH=/authn # For Traefik/gateway routing
replicas: 3
ghcr.io/vishnu2kmohan/keycloak-optimized:26.4.2) is built via GitHub Actions from docker/Dockerfile.keycloak. It pre-compiles Quarkus, enabling --optimized mode and readOnlyRootFilesystem: true for enhanced security.
2. Database Replication
Use PostgreSQL replication for Keycloak database
3. Graceful Degradation
# Keycloak provider gracefully handles failures
try:
result = await keycloak_provider.authenticate(username, password)
except Exception as e:
# Log error but don't crash
logger.error(f"Keycloak authentication failed: {e}")
# Consider fallback to cached credentials or maintenance mode
Compliance
1. GDPR Compliance
- Enable Keycloak audit logging
- Configure user data retention policies
- Implement right-to-be-forgotten workflows
2. SOC 2 Requirements
- Enable MFA in Keycloak
- Implement session timeout policies
- Log all authentication events
3. HIPAA Requirements
- Use encrypted connections (TLS 1.3)
- Implement strong password policies
- Enable comprehensive audit logging
Next Steps
- Complete Setup: Run
make setup-keycloak and configure .env
- Test Integration: Run
python examples/keycloak_usage.py
- Customize Roles: Modify role mapping in
sync_user_to_openfga()
- Production Deploy: Follow Production Deployment Guide
- Monitor & Alert: Set up Grafana dashboards and alerts
Need Help?
