Support Ed25519 Algorithm (#742)

Ed25519 is needed for Arcade AS. This required migrating from
`python-jose` to `joserfc`, because `python-jose` didn't seem to support
Ed25519

libs/arcade-mcp-server/arcade_mcp_server/resource_server/validators/jwks.py

@@ -4,11 +4,16 @@ JWKS-based token validator for MCP Resource Servers.
 Implements OAuth 2.1 Resource Server token validation using JWT with JWKS.
 """
 
+import binascii
 import time
 from typing import Any, cast
 
 import httpx
-from jose import jwk, jwt
+from joserfc import jws, jwt
+from joserfc.errors import JoseError
+from joserfc.jwk import KeySet, KeySetSerialization
+from joserfc.jws import JWSRegistry
+from joserfc.registry import HeaderParameter
 
 from arcade_mcp_server.resource_server.base import (
     AccessTokenValidationOptions,
@@ -19,19 +24,36 @@ from arcade_mcp_server.resource_server.base import (
     TokenExpiredError,
 )
 
-# Note: Only asymmetric algorithms supported
 SUPPORTED_ALGORITHMS = {
+    # RSA
     "RS256",
     "RS384",
     "RS512",
+    # ECDSA
     "ES256",
     "ES384",
     "ES512",
+    # RSA-PSS
     "PS256",
     "PS384",
     "PS512",
+    # EdDSA
+    "Ed25519",
+    "EdDSA",
 }
 
+# EdDSA algorithm aliases
+EDDSA_ALGORITHMS = {"Ed25519", "EdDSA"}
+
+# Custom JWS registry that allows additional header params beyond joserfc's default
+_ACCESS_TOKEN_REGISTRY = JWSRegistry(
+    header_registry={
+        "iss": HeaderParameter("Issuer", "str"),
+        "aud": HeaderParameter("Audience", "str"),
+    },
+    algorithms=list(SUPPORTED_ALGORITHMS),
+)
+
 
 class JWKSTokenValidator(ResourceServerValidator):
     """JWKS-based JWT token validator for simple, explicit token validation.
@@ -114,6 +136,31 @@ class JWKSTokenValidator(ResourceServerValidator):
         self._jwks_cache: dict[str, Any] | None = None
         self._cache_timestamp: float = 0
 
+    def _normalize_algorithm(self, alg: str) -> str:
+        """Normalize algorithm name for comparison.
+
+        EdDSA has multiple names (EdDSA, Ed25519) that should be treated as equivalent.
+
+        Args:
+            alg: Algorithm name
+
+        Returns:
+            Normalized algorithm name
+        """
+        return "Ed25519" if alg in EDDSA_ALGORITHMS else alg
+
+    def _algorithms_match(self, alg1: str, alg2: str) -> bool:
+        """Check if two algorithm names match (considering EdDSA aliases).
+
+        Args:
+            alg1: First algorithm name
+            alg2: Second algorithm name
+
+        Returns:
+            True if algorithms match
+        """
+        return self._normalize_algorithm(alg1) == self._normalize_algorithm(alg2)
+
     async def _fetch_jwks(self) -> dict[str, Any]:
         """Fetch JWKS with caching.
 
@@ -139,6 +186,25 @@ class JWKSTokenValidator(ResourceServerValidator):
         else:
             return self._jwks_cache
 
+    def _get_headers_without_verification(self, token: str) -> dict[str, Any]:
+        """Extract header from JWT without verification.
+
+        Args:
+            token: JWT token string
+
+        Returns:
+            Header dictionary containing 'alg', 'kid', etc.
+
+        Raises:
+            InvalidTokenError: If token format is invalid
+        """
+        try:
+            # Use joserfc's extract_compact to parse JWT without verification
+            obj = jws.extract_compact(token.encode())
+            return obj.headers()
+        except (JoseError, ValueError, binascii.Error) as e:
+            raise InvalidTokenError(f"Invalid JWT format: {e}") from e
+
     def _find_signing_key(self, jwks: dict[str, Any], token: str) -> Any:
         """Find the signing key from JWKS that matches the token's kid.
 
@@ -147,74 +213,92 @@ class JWKSTokenValidator(ResourceServerValidator):
             token: JWT token
 
         Returns:
-            Signing key in PEM format
+            Key object from joserfc KeySet
 
         Raises:
             InvalidTokenError: If no matching key found or algorithm mismatch
         """
-        unverified_header = jwt.get_unverified_header(token)
+        header = self._get_headers_without_verification(token)
-        kid = unverified_header.get("kid")
+        kid = header.get("kid")
-        token_alg = unverified_header.get("alg")
+        token_alg = header.get("alg")
 
         # Validate token algorithm matches configuration (prevent algorithm confusion)
-        if token_alg and token_alg != self.algorithm:
+        if token_alg and not self._algorithms_match(token_alg, self.algorithm):
             raise InvalidTokenError(
                 f"Token algorithm '{token_alg}' doesn't match "
                 f"configured algorithm '{self.algorithm}'"
             )
 
-        for key_data in jwks.get("keys", []):
+        try:
-            if key_data.get("kid") == kid:
+            key_set = KeySet.import_key_set(cast(KeySetSerialization, jwks))
-                key_alg = key_data.get("alg")
+        except Exception as e:
+            raise InvalidTokenError(f"Failed to import JWKS: {e}") from e
 
-                if key_alg and key_alg != self.algorithm:
+        # Find key by kid
+        for key in key_set.keys:
+            if key.kid == kid:
+                key_alg = key.alg
+                if key_alg and not self._algorithms_match(key_alg, self.algorithm):
                     raise InvalidTokenError(
                         f"Key algorithm '{key_alg}' doesn't match "
                         f"configured algorithm '{self.algorithm}'"
                     )
-
+                return key
-                key_obj = jwk.construct(key_data, algorithm=self.algorithm)
-                return key_obj.to_pem().decode("utf-8")
 
         raise InvalidTokenError("No matching key found in JWKS")
 
-    def _decode_token(self, token: str, signing_key: str) -> dict[str, Any]:
+    def _decode_token(self, token: str, signing_key: Any) -> dict[str, Any]:
         """Decode and verify the provided JWT token.
 
+        Uses jwt.decode for signature verification and payload parsing, then
+        performs time-based claims validation manually for leeway handling
+
         Args:
             token: JWT token
-            signing_key: Public key in PEM format
+            signing_key: Key object from joserfc
 
         Returns:
             Decoded token claims
 
         Raises:
-            jwt.ExpiredSignatureError: Token has expired
+            TokenExpiredError: Token has expired
-            jwt.JWTClaimsError: Token claims validation failed (audience/issuer mismatch)
+            InvalidTokenError: Token validation failed
-            jwt.JWTError: Token is invalid
         """
-        decode_options = {
+        if self.algorithm in EDDSA_ALGORITHMS:
-            "verify_signature": True,  # Always verify signature. Cannot be disabled.
+            algorithms = list(EDDSA_ALGORITHMS)
-            "verify_exp": self.validation_options.verify_exp,
+        else:
-            "verify_iat": self.validation_options.verify_iat,
+            algorithms = [self.algorithm]
-            "verify_nbf": self.validation_options.verify_nbf,
-            "verify_aud": False,  # Manual validation for multi-audience support
-            "verify_iss": False,  # Manual validation for multi-issuer support
-            "leeway": self.validation_options.leeway,
-        }
 
-        # Decode token once without aud/iss validation
+        # First, verify signature & decode payload
-        decoded = cast(
+        try:
-            dict[str, Any],
+            result = jwt.decode(
-            jwt.decode(
+                token, signing_key, algorithms=algorithms, registry=_ACCESS_TOKEN_REGISTRY
-                token,
+            )
-                signing_key,
+        except JoseError as e:
-                algorithms=[self.algorithm],
+            raise InvalidTokenError(f"Token signature verification failed: {e}") from e
-                options=decode_options,
-            ),
-        )
 
-        # Manually validate issuer (if flag is enabled)
+        decoded = result.claims
+
+        # Validate time based claims with leeway support
+        current_time = int(time.time())
+        leeway = self.validation_options.leeway
+
+        if self.validation_options.verify_exp:
+            exp = decoded.get("exp")
+            if exp is not None and exp + leeway < current_time:
+                raise TokenExpiredError("Token has expired")
+
+        if self.validation_options.verify_iat:
+            iat = decoded.get("iat")
+            if iat is not None and iat - leeway > current_time:
+                raise InvalidTokenError("Token issued in the future")
+
+        if self.validation_options.verify_nbf:
+            nbf = decoded.get("nbf")
+            if nbf is not None and nbf - leeway > current_time:
+                raise InvalidTokenError("Token not yet valid")
+
+        # Manually validate issuer (if enabled)
         if self.validation_options.verify_iss:
             token_iss = decoded.get("iss")
             if isinstance(self.issuer, list):
@@ -313,12 +397,6 @@ class JWKSTokenValidator(ResourceServerValidator):
                 claims=decoded,
             )
 
-        except jwt.ExpiredSignatureError as e:
-            raise TokenExpiredError("Token has expired") from e
-        except jwt.JWTClaimsError as e:
-            raise InvalidTokenError(f"Token claims validation failed: {e}") from e
-        except jwt.JWTError as e:
-            raise InvalidTokenError(f"Invalid token: {e}") from e
         except (InvalidTokenError, TokenExpiredError):
             raise
         except Exception as e:

libs/arcade-mcp-server/pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "arcade-mcp-server"
-version = "1.14.2"
+version = "1.15.0"
 description = "Model Context Protocol (MCP) server framework for Arcade.dev"
 readme = "README.md"
 authors = [{ name = "Arcade.dev" }]
@@ -34,7 +34,7 @@ dependencies = [
     "anyio>=4.0.0",
     "python-dotenv>=1.0.0",
     "pydantic-settings>=2.10.1",
-    "python-jose[cryptography]>=3.3.0,<4.0.0",
+    "joserfc>=1.5.0",
     "httpx>=0.27.0,<1.0.0",
 ]
 

libs/tests/arcade_mcp_server/test_resource_server_auth.py

@@ -19,7 +19,9 @@ from arcade_mcp_server.resource_server.middleware import ResourceServerMiddlewar
 from arcade_mcp_server.worker import create_arcade_mcp
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
-from jose import jwt
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from joserfc import jwt
+from joserfc.jwk import OKPKey, RSAKey
 
 
 # Test fixtures
@@ -49,11 +51,23 @@ def rsa_keypair():
     return private_key, public_key
 
 
+@pytest.fixture
+def rsa_joserfc_key(rsa_keypair):
+    """Generate joserfc RSAKey from keypair."""
+    private_key, _ = rsa_keypair
+    pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
+    )
+    return RSAKey.import_key(pem)
+
+
 @pytest.fixture
 def serialized_private_key(rsa_keypair):
     """Generate private key as PEM format for testing."""
     private_key, _ = rsa_keypair
-    # Serialize private key to PEM format for python-jose
+    # Serialize private key to PEM format
     pem = private_key.private_bytes(
         encoding=serialization.Encoding.PEM,
         format=serialization.PrivateFormat.PKCS8,
@@ -94,11 +108,9 @@ def jwks_data(rsa_keypair):
 
 
 @pytest.fixture
-def valid_jwt_token(rsa_keypair):
+def valid_jwt_token(rsa_joserfc_key):
     """Generate valid JWT token for testing."""
-    private_key, _ = rsa_keypair
+    claims = {
-
-    payload = {
         "sub": "user123",
         "email": "user@example.com",
         "iss": "https://auth.example.com",
@@ -107,22 +119,16 @@ def valid_jwt_token(rsa_keypair):
         "iat": int(time.time()),
     }
 
-    token = jwt.encode(
+    header = {"alg": "RS256", "kid": "test-key-1"}
-        payload,
+    token = jwt.encode(header, claims, rsa_joserfc_key)
-        private_key,
-        algorithm="RS256",
-        headers={"kid": "test-key-1"},
-    )
 
     return token
 
 
 @pytest.fixture
-def expired_jwt_token(rsa_keypair):
+def expired_jwt_token(rsa_joserfc_key):
     """Generate expired JWT token for testing."""
-    private_key, _ = rsa_keypair
+    claims = {
-
-    payload = {
         "sub": "user123",
         "email": "user@example.com",
         "iss": "https://auth.example.com",
@@ -131,12 +137,92 @@ def expired_jwt_token(rsa_keypair):
         "iat": int(time.time()) - 7200,
     }
 
-    token = jwt.encode(
+    header = {"alg": "RS256", "kid": "test-key-1"}
-        payload,
+    token = jwt.encode(header, claims, rsa_joserfc_key)
-        private_key,
+
-        algorithm="RS256",
+    return token
-        headers={"kid": "test-key-1"},
+
+
+# Ed25519 fixtures
+@pytest.fixture
+def ed25519_keypair():
+    """Generate Ed25519 key pair for testing."""
+    private_key = Ed25519PrivateKey.generate()
+    public_key = private_key.public_key()
+    return private_key, public_key
+
+
+@pytest.fixture
+def ed25519_joserfc_key(ed25519_keypair):
+    """Generate joserfc OKPKey from Ed25519 keypair."""
+    private_key, _ = ed25519_keypair
+    pem = private_key.private_bytes(
+        encoding=serialization.Encoding.PEM,
+        format=serialization.PrivateFormat.PKCS8,
+        encryption_algorithm=serialization.NoEncryption(),
     )
+    return OKPKey.import_key(pem)
+
+
+@pytest.fixture
+def ed25519_jwks_data(ed25519_keypair):
+    """Generate Ed25519 JWKS data for testing."""
+    _, public_key = ed25519_keypair
+
+    # Get the raw public key bytes
+    public_bytes = public_key.public_bytes_raw()
+
+    # Base64url encode the public key
+    x_b64 = base64.urlsafe_b64encode(public_bytes).decode("utf-8").rstrip("=")
+
+    return {
+        "keys": [
+            {
+                "kty": "OKP",
+                "kid": "ed25519-key-1",
+                "use": "sig",
+                "alg": "Ed25519",
+                "crv": "Ed25519",
+                "x": x_b64,
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def valid_ed25519_token(ed25519_joserfc_key):
+    """Generate valid Ed25519 JWT token for testing."""
+    claims = {
+        "sub": "user456",
+        "email": "ed25519user@example.com",
+        "iss": "https://cloud.arcade.dev/oauth2",
+        "aud": "urn:arcade:mcp",
+        "exp": int(time.time()) + 3600,
+        "iat": int(time.time()),
+    }
+
+    header = {"alg": "Ed25519", "kid": "ed25519-key-1"}
+    # Ed25519 is not in joserfc's recommended algorithms, so we must explicitly allow it
+    token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"])
+
+    return token
+
+
+@pytest.fixture
+def expired_ed25519_token(ed25519_joserfc_key):
+    """Generate expired Ed25519 JWT token for testing."""
+    claims = {
+        "sub": "user456",
+        "email": "ed25519user@example.com",
+        "iss": "https://cloud.arcade.dev/oauth2",
+        "aud": "urn:arcade:mcp",
+        "exp": int(time.time()) - 3600,
+        "iat": int(time.time()) - 7200,
+    }
+
+    header = {"alg": "Ed25519", "kid": "ed25519-key-1"}
+    # Ed25519 is not in joserfc's recommended algorithms, so we must explicitly allow it
+    token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"])
 
     return token
 
@@ -185,9 +271,9 @@ class TestJWKSTokenValidator:
                 await validator.validate_token(expired_jwt_token)
 
     @pytest.mark.asyncio
-    async def test_validate_wrong_audience(self, serialized_private_key, jwks_data):
+    async def test_validate_wrong_audience(self, rsa_joserfc_key, jwks_data):
         """Test validating token with wrong audience."""
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://wrong-server.com",  # Wrong audience
@@ -195,12 +281,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -218,9 +300,9 @@ class TestJWKSTokenValidator:
                 await validator.validate_token(token)
 
     @pytest.mark.asyncio
-    async def test_validate_wrong_issuer(self, serialized_private_key, jwks_data):
+    async def test_validate_wrong_issuer(self, rsa_joserfc_key, jwks_data):
         """Test validating token with wrong issuer."""
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://wrong-issuer.com",  # Wrong issuer
             "aud": "https://mcp.example.com/mcp",
@@ -228,12 +310,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -251,21 +329,17 @@ class TestJWKSTokenValidator:
                 await validator.validate_token(token)
 
     @pytest.mark.asyncio
-    async def test_validate_missing_sub_claim(self, serialized_private_key, jwks_data):
+    async def test_validate_missing_sub_claim(self, rsa_joserfc_key, jwks_data):
         """Test validating token without sub claim."""
-        payload = {
+        claims = {
             "iss": "https://auth.example.com",
             "aud": "https://mcp.example.com/mcp",
             "exp": int(time.time()) + 3600,
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -307,11 +381,9 @@ class TestJWKSTokenValidator:
             assert mock_get.call_count == 1
 
     @pytest.mark.asyncio
-    async def test_validate_multiple_audiences_single_token_aud(
+    async def test_validate_multiple_audiences_single_token_aud(self, rsa_joserfc_key, jwks_data):
-        self, serialized_private_key, jwks_data
-    ):
         """Test validator with multiple audiences accepts token with matching single aud."""
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://old-mcp.example.com",  # Matches first audience
@@ -319,12 +391,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -342,12 +410,10 @@ class TestJWKSTokenValidator:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_validate_multiple_audiences_list_token_aud(
+    async def test_validate_multiple_audiences_list_token_aud(self, rsa_joserfc_key, jwks_data):
-        self, serialized_private_key, jwks_data
-    ):
         """Test validator with multiple audiences accepts token with list aud."""
         # Token with list of audiences where one matches the validator's accepted audiences
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": ["https://api1.com", "https://new-mcp.example.com"],  # Second matches
@@ -355,12 +421,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -378,10 +440,10 @@ class TestJWKSTokenValidator:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_validate_multiple_audiences_no_match(self, serialized_private_key, jwks_data):
+    async def test_validate_multiple_audiences_no_match(self, rsa_joserfc_key, jwks_data):
         """Test validator with multiple audiences rejects token with non-matching aud."""
         # Token with audience that doesn't match any of validator's accepted audiences
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://different-server.com",  # Doesn't match
@@ -389,12 +451,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -412,12 +470,10 @@ class TestJWKSTokenValidator:
                 await validator.validate_token(token)
 
     @pytest.mark.asyncio
-    async def test_validate_single_audience_with_list_token_aud(
+    async def test_validate_single_audience_with_list_token_aud(self, rsa_joserfc_key, jwks_data):
-        self, serialized_private_key, jwks_data
-    ):
         """Test validator with single audience accepts token with list aud containing match."""
         # Token with list of audiences where one matches validator's single audience
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": ["https://api1.com", "https://mcp.example.com/mcp", "https://api2.com"],
@@ -425,12 +481,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -448,10 +500,10 @@ class TestJWKSTokenValidator:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_validate_multiple_issuers_efficient(self, serialized_private_key, jwks_data):
+    async def test_validate_multiple_issuers_efficient(self, rsa_joserfc_key, jwks_data):
         """Test that multi-issuer validation is efficient (single decode)."""
         # Token from second issuer in list
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth2.example.com",  # Second in list
             "aud": "https://mcp.example.com/mcp",
@@ -459,12 +511,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -472,9 +520,9 @@ class TestJWKSTokenValidator:
             mock_response.raise_for_status = Mock()
             mock_get.return_value = mock_response
 
-            # Patch jwt.decode to count calls
             with patch(
-                "arcade_mcp_server.resource_server.validators.jwks.jwt.decode", wraps=jwt.decode
+                "arcade_mcp_server.resource_server.validators.jwks.jwt.decode",
+                wraps=jwt.decode,
             ) as mock_decode:
                 validator = JWKSTokenValidator(
                     jwks_uri="https://auth.example.com/.well-known/jwks.json",
@@ -493,9 +541,9 @@ class TestJWKSTokenValidator:
                 assert mock_decode.call_count == 1
 
     @pytest.mark.asyncio
-    async def test_validate_nbf_claim_before_time(self, serialized_private_key, jwks_data):
+    async def test_validate_nbf_claim_before_time(self, rsa_joserfc_key, jwks_data):
         """Test that token with nbf claim in the future is rejected."""
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://mcp.example.com/mcp",
@@ -504,12 +552,8 @@ class TestJWKSTokenValidator:
             "nbf": int(time.time()) + 3600,  # Not valid for 1 hour
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -528,9 +572,9 @@ class TestJWKSTokenValidator:
                 await validator.validate_token(token)
 
     @pytest.mark.asyncio
-    async def test_validate_nbf_claim_disabled(self, serialized_private_key, jwks_data):
+    async def test_validate_nbf_claim_disabled(self, rsa_joserfc_key, jwks_data):
         """Test that token with nbf in future is accepted when verify_nbf=False."""
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://mcp.example.com/mcp",
@@ -539,12 +583,8 @@ class TestJWKSTokenValidator:
             "nbf": int(time.time()) + 3600,  # Not valid for 1 hour
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -564,10 +604,10 @@ class TestJWKSTokenValidator:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_validate_with_leeway(self, serialized_private_key, jwks_data):
+    async def test_validate_with_leeway(self, rsa_joserfc_key, jwks_data):
         """Test that leeway allows slightly expired tokens."""
         # Token expired 30 seconds ago
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://mcp.example.com/mcp",
@@ -575,12 +615,8 @@ class TestJWKSTokenValidator:
             "iat": int(time.time()) - 3600,
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            serialized_private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -600,6 +636,209 @@ class TestJWKSTokenValidator:
             assert user.user_id == "user123"
 
 
+class TestJWKSTokenValidatorEd25519:
+    """Tests for JWKSTokenValidator with Ed25519 algorithm."""
+
+    @pytest.mark.asyncio
+    async def test_validate_valid_ed25519_token(self, valid_ed25519_token, ed25519_jwks_data):
+        """Test validating a valid Ed25519 JWT token."""
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            validator = JWKSTokenValidator(
+                jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                issuer="https://cloud.arcade.dev/oauth2",
+                audience="urn:arcade:mcp",
+                algorithm="Ed25519",
+            )
+
+            user = await validator.validate_token(valid_ed25519_token)
+
+            assert isinstance(user, ResourceOwner)
+            assert user.user_id == "user456"
+            assert user.email == "ed25519user@example.com"
+            assert user.claims["iss"] == "https://cloud.arcade.dev/oauth2"
+
+    @pytest.mark.asyncio
+    async def test_validate_expired_ed25519_token(self, expired_ed25519_token, ed25519_jwks_data):
+        """Test validating an expired Ed25519 JWT token."""
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            validator = JWKSTokenValidator(
+                jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                issuer="https://cloud.arcade.dev/oauth2",
+                audience="urn:arcade:mcp",
+                algorithm="Ed25519",
+            )
+
+            with pytest.raises(TokenExpiredError):
+                await validator.validate_token(expired_ed25519_token)
+
+    @pytest.mark.asyncio
+    async def test_ed25519_algorithm_mismatch(self, valid_jwt_token, ed25519_jwks_data):
+        """Test that RS256 token is rejected when validator expects Ed25519."""
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            validator = JWKSTokenValidator(
+                jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                issuer="https://auth.example.com",
+                audience="https://mcp.example.com/mcp",
+                algorithm="Ed25519",
+            )
+
+            # RS256 token should be rejected when Ed25519 is expected
+            with pytest.raises(InvalidTokenError, match="algorithm"):
+                await validator.validate_token(valid_jwt_token)
+
+    @pytest.mark.asyncio
+    async def test_ed25519_wrong_audience(self, ed25519_joserfc_key, ed25519_jwks_data):
+        """Test Ed25519 token with wrong audience is rejected."""
+        claims = {
+            "sub": "user456",
+            "iss": "https://cloud.arcade.dev/oauth2",
+            "aud": "wrong:audience",
+            "exp": int(time.time()) + 3600,
+            "iat": int(time.time()),
+        }
+
+        header = {"alg": "Ed25519", "kid": "ed25519-key-1"}
+        token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"])
+
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            validator = JWKSTokenValidator(
+                jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                issuer="https://cloud.arcade.dev/oauth2",
+                audience="urn:arcade:mcp",
+                algorithm="Ed25519",
+            )
+
+            with pytest.raises(InvalidTokenError, match="audience"):
+                await validator.validate_token(token)
+
+    @pytest.mark.asyncio
+    async def test_eddsa_algorithm_alias(self, ed25519_joserfc_key, ed25519_jwks_data):
+        """Test that EdDSA algorithm alias works for Ed25519."""
+        claims = {
+            "sub": "user456",
+            "email": "ed25519user@example.com",
+            "iss": "https://cloud.arcade.dev/oauth2",
+            "aud": "urn:arcade:mcp",
+            "exp": int(time.time()) + 3600,
+            "iat": int(time.time()),
+        }
+
+        header = {"alg": "Ed25519", "kid": "ed25519-key-1"}
+        token = jwt.encode(header, claims, ed25519_joserfc_key, algorithms=["Ed25519"])
+
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            # Use EdDSA alias
+            validator = JWKSTokenValidator(
+                jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                issuer="https://cloud.arcade.dev/oauth2",
+                audience="urn:arcade:mcp",
+                algorithm="EdDSA",  # Using EdDSA alias
+            )
+
+            user = await validator.validate_token(token)
+            assert user.user_id == "user456"
+
+    def test_ed25519_supported_algorithm(self):
+        """Test that Ed25519 is in supported algorithms."""
+        # Should not raise
+        validator = JWKSTokenValidator(
+            jwks_uri="https://example.com/jwks",
+            issuer="https://example.com",
+            audience="https://example.com",
+            algorithm="Ed25519",
+        )
+        assert validator.algorithm == "Ed25519"
+
+    def test_eddsa_supported_algorithm(self):
+        """Test that EdDSA is in supported algorithms."""
+        # Should not raise
+        validator = JWKSTokenValidator(
+            jwks_uri="https://example.com/jwks",
+            issuer="https://example.com",
+            audience="https://example.com",
+            algorithm="EdDSA",
+        )
+        assert validator.algorithm == "EdDSA"
+
+
+class TestArcadeASConfiguration:
+    """Tests for Arcade AS configuration."""
+
+    @pytest.mark.asyncio
+    async def test_arcade_as_config(self, valid_ed25519_token, ed25519_jwks_data):
+        """Test configuration matching Arcade AS."""
+        with patch("httpx.AsyncClient.get") as mock_get:
+            mock_response = Mock()
+            mock_response.json.return_value = ed25519_jwks_data
+            mock_response.raise_for_status = Mock()
+            mock_get.return_value = mock_response
+
+            # Configuration matching Arcade AS
+            resource_server_auth = ResourceServerAuth(
+                canonical_url="https://gateway-manager.arcade.dev/mcp",
+                authorization_servers=[
+                    AuthorizationServerEntry(
+                        authorization_server_url="https://cloud.arcade.dev/oauth2",
+                        issuer="https://cloud.arcade.dev/oauth2",
+                        jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                        algorithm="Ed25519",
+                        expected_audiences=[
+                            "urn:arcade:mcp",
+                            "https://gateway-manager.arcade.dev/mcp",
+                        ],
+                    )
+                ],
+            )
+
+            user = await resource_server_auth.validate_token(valid_ed25519_token)
+            assert user.user_id == "user456"
+            assert user.email == "ed25519user@example.com"
+
+    def test_arcade_as_metadata(self):
+        """Test OAuth metadata for Arcade AS configuration."""
+        resource_server_auth = ResourceServerAuth(
+            canonical_url="https://gateway-manager.arcade.dev/mcp",
+            authorization_servers=[
+                AuthorizationServerEntry(
+                    authorization_server_url="https://cloud.arcade.dev/oauth2",
+                    issuer="https://cloud.arcade.dev/oauth2",
+                    jwks_uri="https://cloud.arcade.dev/.well-known/jwks/oauth2",
+                    algorithm="Ed25519",
+                    expected_audiences=["urn:arcade:mcp"],
+                )
+            ],
+        )
+
+        metadata = resource_server_auth.get_resource_metadata()
+        assert metadata["resource"] == "https://gateway-manager.arcade.dev/mcp"
+        assert metadata["authorization_servers"] == ["https://cloud.arcade.dev/oauth2"]
+
+
 # ResourceServerAuth Tests
 class TestResourceServerAuth:
     """Tests for ResourceServerAuth class."""
@@ -639,12 +878,10 @@ class TestResourceServerAuth:
         assert metadata["bearer_methods_supported"] == ["header"]
 
     @pytest.mark.asyncio
-    async def test_expected_audiences_override(self, rsa_keypair, jwks_data):
+    async def test_expected_audiences_override(self, rsa_keypair, jwks_data, rsa_joserfc_key):
         """Test that expected_audiences overrides canonical_url for audience validation."""
-        private_key, _ = rsa_keypair
-
         # Token with custom audience
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "my-authkit-client-id",
@@ -652,12 +889,8 @@ class TestResourceServerAuth:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -681,12 +914,12 @@ class TestResourceServerAuth:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_expected_audiences_multiple_values(self, rsa_keypair, jwks_data):
+    async def test_expected_audiences_multiple_values(
+        self, rsa_keypair, jwks_data, rsa_joserfc_key
+    ):
         """Test that multiple expected_audiences work correctly."""
-        private_key, _ = rsa_keypair
-
         # Token with one of the expected audiences
-        payload = {
+        claims = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "secondary-client-id",
@@ -694,12 +927,8 @@ class TestResourceServerAuth:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -727,11 +956,11 @@ class TestResourceServerAuth:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_expected_audiences_defaults_to_canonical_url(self, rsa_keypair, jwks_data):
+    async def test_expected_audiences_defaults_to_canonical_url(
+        self, rsa_keypair, jwks_data, rsa_joserfc_key
+    ):
         """Test that without expected_audiences, canonical_url is used for audience validation."""
-        private_key, _ = rsa_keypair
+        claims = {
-
-        payload = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "https://mcp.example.com/mcp",
@@ -739,12 +968,8 @@ class TestResourceServerAuth:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -767,11 +992,11 @@ class TestResourceServerAuth:
             assert user.user_id == "user123"
 
     @pytest.mark.asyncio
-    async def test_expected_audiences_wrong_audience_rejected(self, rsa_keypair, jwks_data):
+    async def test_expected_audiences_wrong_audience_rejected(
+        self, rsa_keypair, jwks_data, rsa_joserfc_key
+    ):
         """Test that tokens with wrong audience are rejected even with expected_audiences."""
-        private_key, _ = rsa_keypair
+        claims = {
-
-        payload = {
             "sub": "user123",
             "iss": "https://auth.example.com",
             "aud": "wrong-client-id",  # Not in expected_audiences list
@@ -779,12 +1004,8 @@ class TestResourceServerAuth:
             "iat": int(time.time()),
         }
 
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -1085,11 +1306,11 @@ class TestMultipleAuthorizationServers:
             assert user.email == "user@example.com"
 
     @pytest.mark.asyncio
-    async def test_resource_server_multiple_as_different_jwks(self, rsa_keypair, jwks_data):
+    async def test_resource_server_multiple_as_different_jwks(
+        self, rsa_keypair, jwks_data, rsa_joserfc_key
+    ):
         """Test multiple AS with different JWKS (multi-IdP)."""
-        private_key, _ = rsa_keypair
+        claims1 = {
-
-        payload1 = {
             "sub": "user123",
             "email": "user@workos.com",
             "iss": "https://workos.authkit.app",
@@ -1097,14 +1318,10 @@ class TestMultipleAuthorizationServers:
             "exp": int(time.time()) + 3600,
             "iat": int(time.time()),
         }
-        token1 = jwt.encode(
+        header1 = {"alg": "RS256", "kid": "test-key-1"}
-            payload1,
+        token1 = jwt.encode(header1, claims1, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
-        payload2 = {
+        claims2 = {
             "sub": "user456",
             "email": "user@keycloak.com",
             "iss": "http://localhost:8080/realms/mcp-test",
@@ -1112,12 +1329,8 @@ class TestMultipleAuthorizationServers:
             "exp": int(time.time()) + 3600,
             "iat": int(time.time()),
         }
-        token2 = jwt.encode(
+        header2 = {"alg": "RS256", "kid": "test-key-1"}
-            payload2,
+        token2 = jwt.encode(header2, claims2, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()
@@ -1159,11 +1372,11 @@ class TestMultipleAuthorizationServers:
             assert user2.email == "user@keycloak.com"
 
     @pytest.mark.asyncio
-    async def test_resource_server_rejects_unconfigured_as(self, rsa_keypair, jwks_data):
+    async def test_resource_server_rejects_unconfigured_as(
+        self, rsa_keypair, jwks_data, rsa_joserfc_key
+    ):
         """Test that tokens from unlisted AS are rejected."""
-        private_key, _ = rsa_keypair
+        claims = {
-
-        payload = {
             "sub": "user123",
             "email": "user@evil.com",
             "iss": "https://evil.com",  # Not in configured list (unauthorized issuer)
@@ -1171,12 +1384,8 @@ class TestMultipleAuthorizationServers:
             "exp": int(time.time()) + 3600,
             "iat": int(time.time()),
         }
-        token = jwt.encode(
+        header = {"alg": "RS256", "kid": "test-key-1"}
-            payload,
+        token = jwt.encode(header, claims, rsa_joserfc_key)
-            private_key,
-            algorithm="RS256",
-            headers={"kid": "test-key-1"},
-        )
 
         with patch("httpx.AsyncClient.get") as mock_get:
             mock_response = Mock()