Validate Casdoor tokens in your MCP server

This guide contains complete MCP servers in Python, Node.js, and Go that use Casdoor as their authorization server. Each one publishes Protected Resource Metadata, validates tokens, and checks scopes.

---

Learning outcomes

• Run an MCP server that validates Casdoor tokens.
• Connect it to Claude Desktop.
• Prepare the server for production.

What you need

• Casdoor configured for your MCP server. See Set up Casdoor for your MCP server.
• An MCP client for testing, such as Claude Desktop

---

uwaga

To keep the examples short, some of them comment out parts of the token validation. Before you deploy a server, turn the validation on and validate every request.

Python

The example uses the official mcp SDK and PyJWT.

1. Install the packages:

   pip install mcp PyJWT cryptography requests

2. Save the following code as mcp_server.py:

   """
   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())

3. Add the server to the Claude Desktop configuration, on macOS ~/Library/Application Support/Claude/claude_desktop_config.json:

   {
     "mcpServers": {
       "example-files": {
         "command": "python",
         "args": ["/path/to/mcp_server.py"]
       }
     }
   }

4. Restart Claude Desktop and check that the tools appear.

Node.js

The example uses the official @modelcontextprotocol/sdk and jose.

1. Install the packages:

   npm install @modelcontextprotocol/sdk jose

2. Save the following code as mcp-server.js:

   /**
    * 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);

3. Add "type": "module" to package.json:

   {
     "name": "mcp-example-server",
     "version": "1.0.0",
     "type": "module",
     "dependencies": {
       "@modelcontextprotocol/sdk": "^0.5.0",
       "jose": "^5.0.0"
     }
   }

4. Add the server to the Claude Desktop configuration:

   {
     "mcpServers": {
       "example-files": {
         "command": "node",
         "args": ["/path/to/mcp-server.js"]
       }
     }
   }

5. Restart Claude Desktop.

Go

The example uses the Casdoor Go SDK and golang-jwt.

1. Install the packages:

   go get github.com/casdoor/casdoor-go-sdk/casdoorsdk
   go get github.com/golang-jwt/jwt/v5

2. Save the following code as main.go:

   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))
   }

3. Run the server:

   go run main.go

4. Request the metadata document:

   curl http://localhost:8080/.well-known/oauth-protected-resource

Prepare for production

Security

• Validate every request: Turn on all checks that the examples comment out.
• Use HTTPS: A token sent over HTTP can be intercepted.
• Limit the rate: Limit the requests per token or per user.
• Log failures: Record failed authorizations for monitoring.

Errors

Tell clients why authorization failed, for example:

{
  "error": "insufficient_scope",
  "error_description": "Token missing required scope: files:write",
  "scope": "files:read files:write"
}

Performance

• Cache the JWKS: Fetching the keys for each request is slow. Cache them and refresh the cache periodically, for example every 24 hours.
• Reuse validation results: Validate a token once per session where your design allows it.
• Reuse connections: Use connection pooling for the requests to the JWKS endpoint.

Troubleshooting

The server can't fetch the JWKS

Check the URL of the JWKS endpoint, the network connection, and the firewall rules.

Tokens are reported as expired

Synchronize the clock of your server. JWT expiry is checked against the current time.

Tools run without the required scope

Check that the scope checks aren't commented out.

Contribute an example

If you built an MCP server with Casdoor in another language, such as Rust, Ruby, PHP, Java, or C#, add it to this page: fork the casdoor-website repository, add your example, and open a pull request.

See also

• Set up Casdoor for your MCP server
• MCP authorization specification
• Casdoor Go SDK
• Define custom scopes