0
# Authentication & Security
1
2
OAuth 2.0 authentication, bearer token verification, transport security settings, and middleware for protecting MCP server endpoints. The authentication system provides secure access control for MCP servers with support for various OAuth flows and token verification strategies.
3
4
## Capabilities
5
6
### Authentication Settings
7
8
Configuration classes for OAuth 2.0 authentication and security settings.
9
10
```python { .api }
11
class AuthSettings:
12
def __init__(
13
self,
14
client_id: str,
15
client_secret: str,
16
authorization_endpoint: str,
17
token_endpoint: str,
18
scopes: list[str] | None = None,
19
**kwargs
20
):
21
"""
22
OAuth 2.0 authentication configuration.
23
24
Parameters:
25
- client_id: OAuth client identifier
26
- client_secret: OAuth client secret
27
- authorization_endpoint: OAuth authorization URL
28
- token_endpoint: OAuth token exchange URL
29
- scopes: Required OAuth scopes
30
- **kwargs: Additional OAuth parameters
31
"""
32
33
class ClientRegistrationOptions:
34
def __init__(
35
self,
36
registration_endpoint: str,
37
client_name: str,
38
redirect_uris: list[str],
39
**kwargs
40
):
41
"""
42
OAuth client registration options.
43
44
Parameters:
45
- registration_endpoint: Client registration URL
46
- client_name: Human-readable client name
47
- redirect_uris: List of valid redirect URIs
48
- **kwargs: Additional registration parameters
49
"""
50
51
class RevocationOptions:
52
def __init__(
53
self,
54
revocation_endpoint: str,
55
revocation_endpoint_auth_method: str = "client_secret_basic",
56
**kwargs
57
):
58
"""
59
OAuth token revocation options.
60
61
Parameters:
62
- revocation_endpoint: Token revocation URL
63
- revocation_endpoint_auth_method: Authentication method for revocation
64
- **kwargs: Additional revocation parameters
65
"""
66
```
67
68
### Token Verification
69
70
Interfaces and implementations for verifying OAuth tokens and bearer tokens.
71
72
```python { .api }
73
class TokenVerifier:
74
async def verify_token(self, token: str) -> dict[str, Any]:
75
"""
76
Verify an OAuth token and return claims.
77
78
Parameters:
79
- token: Bearer token to verify
80
81
Returns:
82
Dictionary containing token claims and user information
83
"""
84
85
class ProviderTokenVerifier(TokenVerifier):
86
def __init__(
87
self,
88
auth_settings: AuthSettings,
89
jwks_url: str | None = None,
90
**kwargs
91
):
92
"""
93
Provider-based token verifier using OAuth introspection.
94
95
Parameters:
96
- auth_settings: OAuth authentication settings
97
- jwks_url: JSON Web Key Set URL for token validation
98
- **kwargs: Additional verification options
99
"""
100
101
async def verify_token(self, token: str) -> dict[str, Any]:
102
"""Verify token using OAuth provider introspection."""
103
104
class JWTTokenVerifier(TokenVerifier):
105
def __init__(
106
self,
107
public_key: str,
108
algorithm: str = "RS256",
109
audience: str | None = None,
110
**kwargs
111
):
112
"""
113
JWT token verifier for local token validation.
114
115
Parameters:
116
- public_key: Public key for JWT verification
117
- algorithm: JWT signing algorithm
118
- audience: Expected token audience
119
- **kwargs: Additional JWT validation options
120
"""
121
122
async def verify_token(self, token: str) -> dict[str, Any]:
123
"""Verify JWT token locally."""
124
```
125
126
### Authentication Middleware
127
128
Middleware components for enforcing authentication requirements and managing authentication context.
129
130
```python { .api }
131
class BearerAuthBackend:
132
def __init__(
133
self,
134
token_verifier: TokenVerifier,
135
scheme_name: str = "Bearer",
136
**kwargs
137
):
138
"""
139
Bearer token authentication backend.
140
141
Parameters:
142
- token_verifier: Token verification implementation
143
- scheme_name: Authentication scheme name
144
- **kwargs: Additional backend options
145
"""
146
147
async def authenticate(self, request) -> dict[str, Any] | None:
148
"""
149
Authenticate request using bearer token.
150
151
Parameters:
152
- request: HTTP request object
153
154
Returns:
155
User information dictionary or None if authentication fails
156
"""
157
158
class RequireAuthMiddleware:
159
def __init__(
160
self,
161
auth_backend: BearerAuthBackend,
162
excluded_paths: list[str] | None = None,
163
**kwargs
164
):
165
"""
166
Middleware requiring authentication for protected endpoints.
167
168
Parameters:
169
- auth_backend: Authentication backend
170
- excluded_paths: Paths that don't require authentication
171
- **kwargs: Additional middleware options
172
"""
173
174
class AuthContextMiddleware:
175
def __init__(self, **kwargs):
176
"""
177
Middleware for providing authentication context to handlers.
178
179
Parameters:
180
- **kwargs: Middleware configuration options
181
"""
182
```
183
184
### Transport Security
185
186
Security settings for protecting transport connections and preventing common attacks.
187
188
```python { .api }
189
class TransportSecuritySettings:
190
def __init__(
191
self,
192
dns_rebinding_protection: bool = True,
193
allowed_origins: list[str] | None = None,
194
require_https: bool = True,
195
**kwargs
196
):
197
"""
198
Transport security configuration.
199
200
Parameters:
201
- dns_rebinding_protection: Enable DNS rebinding attack protection
202
- allowed_origins: List of allowed origins for CORS
203
- require_https: Require HTTPS for security-sensitive operations
204
- **kwargs: Additional security options
205
"""
206
```
207
208
## Usage Examples
209
210
### Basic OAuth Authentication
211
212
```python
213
from mcp.server import FastMCP
214
from mcp.server.auth import AuthSettings, ProviderTokenVerifier
215
216
# Configure OAuth authentication
217
auth_settings = AuthSettings(
218
client_id="your-client-id",
219
client_secret="your-client-secret",
220
authorization_endpoint="https://auth.example.com/oauth/authorize",
221
token_endpoint="https://auth.example.com/oauth/token",
222
scopes=["read", "write"]
223
)
224
225
# Create token verifier
226
token_verifier = ProviderTokenVerifier(
227
auth_settings=auth_settings,
228
jwks_url="https://auth.example.com/.well-known/jwks.json"
229
)
230
231
# Create authenticated server
232
app = FastMCP(
233
"secure-server",
234
auth_server_provider=auth_settings,
235
token_verifier=token_verifier
236
)
237
238
@app.tool()
239
async def protected_tool(ctx, data: str) -> str:
240
"""Tool requiring authentication."""
241
user_id = ctx.user_id
242
if not user_id:
243
raise ValueError("Authentication required")
244
245
return f"Processed data for user {user_id}: {data}"
246
247
# Run with authentication
248
if __name__ == "__main__":
249
import asyncio
250
asyncio.run(app.run_sse(port=8443))
251
```
252
253
### JWT Token Verification
254
255
```python
256
from mcp.server import FastMCP, Context
257
from mcp.server.auth import JWTTokenVerifier
258
259
# JWT verification with public key
260
public_key = """-----BEGIN PUBLIC KEY-----
261
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...
262
-----END PUBLIC KEY-----"""
263
264
token_verifier = JWTTokenVerifier(
265
public_key=public_key,
266
algorithm="RS256",
267
audience="mcp-server"
268
)
269
270
app = FastMCP("jwt-server", token_verifier=token_verifier)
271
272
@app.tool()
273
async def user_info(ctx: Context) -> dict:
274
"""Get authenticated user information."""
275
if not ctx.user_id:
276
raise ValueError("JWT authentication required")
277
278
# Access token claims from context
279
return {
280
"user_id": ctx.user_id,
281
"request_id": ctx.request_id,
282
"authenticated": True
283
}
284
285
if __name__ == "__main__":
286
import asyncio
287
asyncio.run(app.run_sse(port=8000))
288
```
289
290
### Custom Authentication Middleware
291
292
```python
293
from mcp.server import FastMCP
294
from mcp.server.auth import BearerAuthBackend, RequireAuthMiddleware, TokenVerifier
295
296
class CustomTokenVerifier(TokenVerifier):
297
"""Custom token verification logic."""
298
299
async def verify_token(self, token: str) -> dict[str, Any]:
300
# Custom token validation
301
if token.startswith("valid_"):
302
user_id = token.replace("valid_", "")
303
return {
304
"user_id": user_id,
305
"roles": ["user"],
306
"permissions": ["read", "write"]
307
}
308
else:
309
raise ValueError("Invalid token")
310
311
# Setup authentication chain
312
token_verifier = CustomTokenVerifier()
313
auth_backend = BearerAuthBackend(token_verifier)
314
auth_middleware = RequireAuthMiddleware(
315
auth_backend,
316
excluded_paths=["/health", "/public"]
317
)
318
319
app = FastMCP("custom-auth-server")
320
321
# Apply middleware
322
app.add_middleware(auth_middleware)
323
324
@app.get("/health")
325
async def health_check():
326
"""Public health check endpoint."""
327
return {"status": "healthy"}
328
329
@app.tool()
330
async def secure_operation(ctx, operation: str) -> str:
331
"""Protected tool requiring authentication."""
332
user_id = ctx.user_id
333
return f"User {user_id} performed: {operation}"
334
335
if __name__ == "__main__":
336
import asyncio
337
asyncio.run(app.run_sse(port=8000))
338
```
339
340
### Transport Security Configuration
341
342
```python
343
from mcp.server import FastMCP
344
from mcp.server.transport_security import TransportSecuritySettings
345
346
# Configure transport security
347
security_settings = TransportSecuritySettings(
348
dns_rebinding_protection=True,
349
allowed_origins=[
350
"https://app.example.com",
351
"https://admin.example.com"
352
],
353
require_https=True
354
)
355
356
app = FastMCP(
357
"secure-transport-server",
358
transport_security=security_settings
359
)
360
361
@app.tool()
362
async def sensitive_data() -> dict:
363
"""Tool handling sensitive data."""
364
return {
365
"data": "sensitive information",
366
"security": "protected by transport security"
367
}
368
369
if __name__ == "__main__":
370
import asyncio
371
# Run on HTTPS with security settings
372
asyncio.run(app.run_sse(
373
host="0.0.0.0",
374
port=8443,
375
ssl_certfile="cert.pem",
376
ssl_keyfile="key.pem"
377
))
378
```
379
380
### Client-Side Authentication
381
382
```python
383
import asyncio
384
from mcp import ClientSession, sse_client
385
from mcp.client.auth import ClientAuth
386
387
async def authenticated_client():
388
# Configure client authentication
389
auth = ClientAuth(token="valid_user123")
390
headers = await auth.get_headers()
391
392
# Connect with authentication headers
393
async with sse_client(
394
"https://localhost:8443/mcp",
395
headers=headers
396
) as (read, write):
397
async with ClientSession(read, write) as session:
398
await session.initialize()
399
400
# Call authenticated tools
401
result = await session.call_tool("secure_operation", {
402
"operation": "data_processing"
403
})
404
print(f"Authenticated result: {result}")
405
406
asyncio.run(authenticated_client())
407
```