第三方 MCP 服务器集成
本页提供使用 Casdoor 做 OAuth 2.1 的 MCP 服务器的完整可运行示例：受保护资源元数据、JWT 校验和 scope 检查。每个示例几分钟就能跑起来。
先决条件
- 配置了 MCP 应用的 Casdoor（配置）
- 用于测试的 MCP 客户端（例如 Claude Desktop）
Python 示例
本示例使用官方的 mcp SDK 和 PyJWT 校验令牌。
安装
pip install mcp PyJWT cryptography requests
完整服务器代码
"""
MCP Server with Casdoor OAuth Authentication
Demonstrates Protected Resource Metadata, JWT validation, and scope enforcement
"""
import asyncio
import json
from typing import Any
import jwt
import requests
from jwt import PyJWKClient
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent
# Configuration - Replace with your Casdoor instance
CASDOOR_URL = "https://your-casdoor.com"
MCP_SERVER_URL = "https://your-mcp-server.com"
JWKS_URL = f"{CASDOOR_URL}/.well-known/jwks"
# Initialize JWKS client for token validation
jwks_client = PyJWKClient(JWKS_URL)
# Create MCP server instance
app = Server("example-mcp-server")
def validate_token(token: str) -> dict:
"""
Validate JWT token from Casdoor using JWKS.
Returns decoded token with claims if valid.
Raises jwt.InvalidTokenError if invalid.
"""
try:
# Get signing key from JWKS
signing_key = jwks_client.get_signing_key_from_jwt(token)
# Verify and decode token
decoded = jwt.decode(
token,
signing_key.key,
algorithms=["RS256"],
audience=MCP_SERVER_URL, # Verify audience matches our server
options={
"verify_signature": True,
"verify_exp": True,
"verify_aud": True,
}
)
return decoded
except jwt.InvalidTokenError as e:
raise ValueError(f"Invalid token: {e}")
def check_scope(token_data: dict, required_scope: str) -> None:
"""
Check if token contains required scope.
Raises PermissionError if scope is missing.
"""
scopes = token_data.get("scope", "").split()
if required_scope not in scopes:
raise PermissionError(f"Missing required scope: {required_scope}")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available tools"""
return [
Tool(
name="read_file",
description="Read contents of a file",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to read"
}
},
"required": ["path"]
}
),
Tool(
name="write_file",
description="Write content to a file",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Path to the file to write"
},
"content": {
"type": "string",
"description": "Content to write to the file"
}
},
"required": ["path", "content"]
}
),
Tool(
name="list_files",
description="List files in a directory",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Directory path to list"
}
},
"required": ["path"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: Any) -> list[TextContent]:
"""
Handle tool calls with OAuth token validation and scope enforcement.
In production, extract token from request context.
For this example, we'll show the validation logic.
"""
# In a real MCP server, extract token from the request headers
# token = request.headers.get("Authorization", "").replace("Bearer ", "")
# For demonstration, we'll skip actual token extraction
# Example token validation (uncomment in production):
# try:
# token_data = validate_token(token)
# except ValueError as e:
# return [TextContent(type="text", text=f"Authentication failed: {e}")]
# Handle each tool with appropriate scope checks
if name == "read_file":
# Requires files:read scope
# check_scope(token_data, "files:read")
path = arguments.get("path")
try:
with open(path, 'r') as f:
content = f.read()
return [TextContent(
type="text",
text=f"File contents:\n{content}"
)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error reading file: {e}"
)]
elif name == "write_file":
# Requires files:write scope
# check_scope(token_data, "files:write")
path = arguments.get("path")
content = arguments.get("content")
try:
with open(path, 'w') as f:
f.write(content)
return [TextContent(
type="text",
text=f"Successfully wrote to {path}"
)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error writing file: {e}"
)]
elif name == "list_files":
# Requires files:list scope
# check_scope(token_data, "files:list")
import os
path = arguments.get("path")
try:
files = os.listdir(path)
return [TextContent(
type="text",
text=f"Files in {path}:\n" + "\n".join(files)
)]
except Exception as e:
return [TextContent(
type="text",
text=f"Error listing files: {e}"
)]
return [TextContent(type="text", text=f"Unknown tool: {name}")]
async def serve_protected_resource_metadata():
"""
Serve Protected Resource Metadata endpoint.
In production, integrate this with your HTTP framework.
"""
metadata = {
"resource": MCP_SERVER_URL,
"authorization_servers": [CASDOOR_URL],
"scopes_supported": [
"files:read",
"files:write",
"files:list"
],
"bearer_methods_supported": ["header"]
}
return metadata
async def main():
"""Run the MCP server"""
async with stdio_server() as (read_stream, write_stream):
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
if __name__ == "__main__":
asyncio.run(main())
测试 Python 服务器
-
把代码 保存 为
mcp_server.py -
配置 Claude Desktop：添加到
~/Library/Application Support/Claude/claude_desktop_config.json（macOS）：
{
"mcpServers": {
"example-files": {
"command": "python",
"args": ["/path/to/mcp_server.py"]
}
}
}
- 重启 Claude Desktop，确认工具出现在 MCP 工具列表中
Node.js 示例
本示例使用官方的 @modelcontextprotocol/sdk 和 jose 校验 JWT。
安装
npm install @modelcontextprotocol/sdk jose
完整服务器代码
/**
* MCP Server with Casdoor OAuth Authentication
* Demonstrates Protected Resource Metadata, JWT validation, and scope enforcement
*/
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { createRemoteJWKSet, jwtVerify } from 'jose';
import {
CallToolRequestSchema,
ListToolsRequestSchema,
} from '@modelcontextprotocol/sdk/types.js';
// Configuration - Replace with your Casdoor instance
const CASDOOR_URL = 'https://your-casdoor.com';
const MCP_SERVER_URL = 'https://your-mcp-server.com';
const JWKS_URL = `${CASDOOR_URL}/.well-known/jwks`;
// Initialize JWKS for token validation
const JWKS = createRemoteJWKSet(new URL(JWKS_URL));
/**
* Validate JWT token from Casdoor using JWKS.
*
* @param {string} token - JWT token to validate
* @returns {Promise<object>} Decoded token payload
* @throws {Error} If token is invalid
*/
async function validateToken(token) {
try {
const { payload } = await jwtVerify(token, JWKS, {
audience: MCP_SERVER_URL,
issuer: CASDOOR_URL,
});
return payload;
} catch (error) {
throw new Error(`Invalid token: ${error.message}`);
}
}
/**
* Check if token contains required scope.
*
* @param {object} tokenData - Decoded token payload
* @param {string} requiredScope - Scope to check for
* @throws {Error} If scope is missing
*/
function checkScope(tokenData, requiredScope) {
const scopes = (tokenData.scope || '').split(' ');
if (!scopes.includes(requiredScope)) {
throw new Error(`Missing required scope: ${requiredScope}`);
}
}
// Create MCP server instance
const server = new Server(
{
name: 'example-mcp-server',
version: '1.0.0',
},
{
capabilities: {
tools: {},
},
}
);
// List available tools
server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: 'read_file',
description: 'Read contents of a file',
inputSchema: {
type: 'object',
properties: {
path: {
type: 'string',
description: 'Path to the file to read',
},
},
required: ['path'],
},
},
{
name: 'write_file',
description: 'Write content to a file',
inputSchema: {
type: 'object',
properties: {
path: {
type: 'string',
description: 'Path to the file to write',
},
content: {
type: 'string',
description: 'Content to write to the file',
},
},
required: ['path', 'content'],
},
},
{
name: 'list_files',
description: 'List files in a directory',
inputSchema: {
type: 'object',
properties: {
path: {
type: 'string',
description: 'Directory path to list',
},
},
required: ['path'],
},
},
],
};
});
// Handle tool calls
server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
// In production, extract token from request context
// const token = request.headers?.authorization?.replace('Bearer ', '');
// const tokenData = await validateToken(token);
try {
switch (name) {
case 'read_file': {
// checkScope(tokenData, 'files:read');
const fs = await import('fs/promises');
const content = await fs.readFile(args.path, 'utf-8');
return {
content: [
{
type: 'text',
text: `File contents:\n${content}`,
},
],
};
}
case 'write_file': {
// checkScope(tokenData, 'files:write');
const fs = await import('fs/promises');
await fs.writeFile(args.path, args.content, 'utf-8');
return {
content: [
{
type: 'text',
text: `Successfully wrote to ${args.path}`,
},
],
};
}
case 'list_files': {
// checkScope(tokenData, 'files:list');
const fs = await import('fs/promises');
const files = await fs.readdir(args.path);
return {
content: [
{
type: 'text',
text: `Files in ${args.path}:\n${files.join('\n')}`,
},
],
};
}
default:
throw new Error(`Unknown tool: ${name}`);
}
} catch (error) {
return {
content: [
{
type: 'text',
text: `Error: ${error.message}`,
},
],
isError: true,
};
}
});
/**
* Protected Resource Metadata
* In production, serve this via HTTP endpoint at /.well-known/oauth-protected-resource
*/
export const protectedResourceMetadata = {
resource: MCP_SERVER_URL,
authorization_servers: [CASDOOR_URL],
scopes_supported: ['files:read', 'files:write', 'files:list'],
bearer_methods_supported: ['header'],
};
// Run the server
async function main() {
const transport = new StdioServerTransport();
await server.connect(transport);
console.error('MCP Server running on stdio');
}
main().catch(console.error);
测试 Node.js 服务器
-
把代码 保存 为
mcp-server.js -
在
package.json中加上"type": "module"：
{
"name": "mcp-example-server",
"version": "1.0.0",
"type": "module",
"dependencies": {
"@modelcontextprotocol/sdk": "^0.5.0",
"jose": "^5.0.0"
}
}
- 配置 Claude Desktop：
{
"mcpServers": {
"example-files": {
"command": "node",
"args": ["/path/to/mcp-server.js"]
}
}
}
- 重启 Claude Desktop
Go 示例
本示例演示如何使用 Casdoor Go SDK 和标准库校验 JWT 进行集成。
安装
go get github.com/casdoor/casdoor-go-sdk/casdoorsdk
go get github.com/golang-jwt/jwt/v5
完整服务器代码
package main
import (
"context"
"encoding/json"
"fmt"
"log"
"net/http"
"os"
"strings"
"github.com/golang-jwt/jwt/v5"
)
// Configuration - Replace with your Casdoor instance
const (
CasdoorURL = "https://your-casdoor.com"
MCPServerURL = "https://your-mcp-server.com"
JWKSEndpoint = CasdoorURL + "/.well-known/jwks"
)
// ProtectedResourceMetadata represents the MCP server's OAuth metadata
type ProtectedResourceMetadata struct {
Resource string `json:"resource"`
AuthorizationServers []string `json:"authorization_servers"`
ScopesSupported []string `json:"scopes_supported"`
BearerMethodsSupported []string `json:"bearer_methods_supported"`
}
// MCPTool represents an MCP tool definition
type MCPTool struct {
Name string `json:"name"`
Description string `json:"description"`
InputSchema interface{} `json:"inputSchema"`
}
// TokenClaims represents JWT claims from Casdoor
type TokenClaims struct {
Scope string `json:"scope"`
jwt.RegisteredClaims
}
// ValidateToken validates a JWT token using Casdoor's JWKS
func ValidateToken(tokenString string) (*TokenClaims, error) {
// Parse token
token, err := jwt.ParseWithClaims(tokenString, &TokenClaims{}, func(token *jwt.Token) (interface{}, error) {
// Verify signing method
if _, ok := token.Method.(*jwt.SigningMethodRSA); !ok {
return nil, fmt.Errorf("unexpected signing method: %v", token.Header["alg"])
}
// In production, fetch JWKS and extract the appropriate key
// For simplicity, this example shows the structure
// Use a library like github.com/lestrrat-go/jwx for full JWKS support
return nil, fmt.Errorf("JWKS key fetching not implemented in this example")
})
if err != nil {
return nil, err
}
claims, ok := token.Claims.(*TokenClaims)
if !ok || !token.Valid {
return nil, fmt.Errorf("invalid token")
}
// Verify audience
if !claims.VerifyAudience(MCPServerURL, true) {
return nil, fmt.Errorf("invalid audience")
}
return claims, nil
}
// CheckScope verifies that the token contains the required scope
func CheckScope(claims *TokenClaims, requiredScope string) error {
scopes := strings.Split(claims.Scope, " ")
for _, scope := range scopes {
if scope == requiredScope {
return nil
}
}
return fmt.Errorf("missing required scope: %s", requiredScope)
}
// ProtectedResourceMetadataHandler serves the Protected Resource Metadata
func ProtectedResourceMetadataHandler(w http.ResponseWriter, r *http.Request) {
metadata := ProtectedResourceMetadata{
Resource: MCPServerURL,
AuthorizationServers: []string{CasdoorURL},
ScopesSupported: []string{"files:read", "files:write", "files:list"},
BearerMethodsSupported: []string{"header"},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(metadata)
}
// ListToolsHandler returns available MCP tools
func ListToolsHandler(w http.ResponseWriter, r *http.Request) {
tools := []MCPTool{
{
Name: "read_file",
Description: "Read contents of a file",
InputSchema: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"path": map[string]string{
"type": "string",
"description": "Path to the file to read",
},
},
"required": []string{"path"},
},
},
{
Name: "write_file",
Description: "Write content to a file",
InputSchema: map[string]interface{}{
"type": "object",
"properties": map[string]interface{}{
"path": map[string]string{
"type": "string",
"description": "Path to the file to write",
},
"content": map[string]string{
"type": "string",
"description": "Content to write",
},
},
"required": []string{"path", "content"},
},
},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(map[string]interface{}{
"tools": tools,
})
}
// CallToolHandler handles MCP tool invocations with scope enforcement
func CallToolHandler(w http.ResponseWriter, r *http.Request) {
// Extract token from Authorization header
authHeader := r.Header.Get("Authorization")
if authHeader == "" {
http.Error(w, "Missing authorization header", http.StatusUnauthorized)
return
}
tokenString := strings.TrimPrefix(authHeader, "Bearer ")
// Validate token (in production, uncomment this)
// claims, err := ValidateToken(tokenString)
// if err != nil {
// http.Error(w, fmt.Sprintf("Invalid token: %v", err), http.StatusUnauthorized)
// return
// }
// Decode request body
var request struct {
Name string `json:"name"`
Arguments map[string]interface{} `json:"arguments"`
}
if err := json.NewDecoder(r.Body).Decode(&request); err != nil {
http.Error(w, "Invalid request", http.StatusBadRequest)
return
}
// Handle tools with scope enforcement
var result string
var err error
switch request.Name {
case "read_file":
// Check scope (in production, uncomment this)
// if err := CheckScope(claims, "files:read"); err != nil {
// http.Error(w, err.Error(), http.StatusForbidden)
// return
// }
path := request.Arguments["path"].(string)
content, err := os.ReadFile(path)
if err != nil {
result = fmt.Sprintf("Error reading file: %v", err)
} else {
result = fmt.Sprintf("File contents:\n%s", content)
}
case "write_file":
// if err := CheckScope(claims, "files:write"); err != nil {
// http.Error(w, err.Error(), http.StatusForbidden)
// return
// }
path := request.Arguments["path"].(string)
content := request.Arguments["content"].(string)
err = os.WriteFile(path, []byte(content), 0644)
if err != nil {
result = fmt.Sprintf("Error writing file: %v", err)
} else {
result = fmt.Sprintf("Successfully wrote to %s", path)
}
default:
http.Error(w, "Unknown tool", http.StatusNotFound)
return
}
response := map[string]interface{}{
"content": []map[string]string{
{
"type": "text",
"text": result,
},
},
}
w.Header().Set("Content-Type", "application/json")
json.NewEncoder(w).Encode(response)
}
func main() {
http.HandleFunc("/.well-known/oauth-protected-resource", ProtectedResourceMetadataHandler)
http.HandleFunc("/tools/list", ListToolsHandler)
http.HandleFunc("/tools/call", CallToolHandler)
log.Println("MCP Server running on :8080")
log.Fatal(http.ListenAndServe(":8080", nil))
}
测试 Go 服务器
-
把代码 保存 为
main.go -
运行服务器：
go run main.go
- 测试受保护资源元数据：
curl http://localhost:8080/.well-known/oauth-protected-resource
生产环境注意事项
安全最佳实践
-
生产环境务必校验令牌：示例为了简洁把校验逻辑注释掉了。生产环境 每个请求都要校验。
-
使用 HTTPS：生产环境的 MCP 服务器必须使用 HTTPS。通过 HTTP 传输的令牌可能被截获。
-
实现限流：按令牌 / 用户限制请求数，防止服务器被滥用。
-
记录授权失败：跟踪失败的授权尝试，用于安全监控。
-
定期刷新 JWKS 缓存：JWKS 可以缓存，但要定期刷新（例如每 24 小时）。
错误处理
返回描述清楚的错误，帮助客户端排查授权问题：
{
"error": "insufficient_scope",
"error_description": "Token missing required scope: files:write",
"scope": "files:read files:write"
}
性能优化
-
缓存 JWKS 响应：每个请求都去取 JWKS 很慢。缓存密钥并定期刷新。
-
每个会话只校验一次令牌：可以的话，校验一次后在会话期间缓存结果。
-
使用连接池：获取 JWKS 时复用 HTTP 连接。
独立仓库示例
完整、可用于生产的示例见以下仓库：
- Python：casdoor/example-mcp-server-python（计划中）
- Node.js：casdoor/example-mcp-server-nodejs（计划中）
- Go：casdoor/example-mcp-server-go（计划中）
下一步
- 配置指南 →：回顾 Casdoor 和 MCP 服务器的配置
- MCP 规范 →：阅读官方授权规范
- Casdoor Go SDK →：使用 Casdoor 官方 Go SDK，集成更简单
- 自定义 Scope →：进一步了解如何定义细粒度权限
故障排除
"JWKS fetch failed"
问题：服务器取不到 Casdoor 的 JWKS 接口。
解决：确认 JWKS 地址正确且可以访问。检查网络连通性和防火墙规则。
"Token expired"
问题：JWT 令牌已过期。
解决：确保服务器系统时钟已同步。 JWT 包含过期时间戳，会与当前时间比较校验。
"Scope enforcement not working"
问题：没有所需 scope 工具也能执行。
解决：确认生产环境中的 scope 检查逻辑 没有 被注释掉。示例为了演示把校验注释掉了。
贡献
如果你开发了使用 Casdoor 认证的 MCP 服务器，欢迎贡献你的示例：
- Fork Casdoor 文档网站仓库
- 把你的示例加到本页
- 提交 pull request
我们特别欢迎以下语言的示例：
- Rust
- Ruby
- PHP
- Java
- C#