update token + revoke to use form data

Author: praboud-ant

CLAUDE.md

@@ -104,6 +104,10 @@ This document contains critical information about working with this codebase. Fo
      - Add None checks
      - Narrow string types
      - Match existing patterns
+   - Pytest:
+     - If the tests aren't finding the anyio pytest mark, try adding PYTEST_DISABLE_PLUGIN_AUTOLOAD=""
+       to the start of the pytest run command eg:
+       `PYTEST_DISABLE_PLUGIN_AUTOLOAD="" uv run --frozen pytest`
 
 3. Best Practices
    - Check git status before commits

pyproject.toml

@@ -30,6 +30,7 @@ dependencies = [
     "sse-starlette>=1.6.1",
     "pydantic-settings>=2.5.2",
     "uvicorn>=0.23.1",
+    "python-multipart",
 ]
 
 [project.optional-dependencies]
@@ -71,9 +72,6 @@ strict = [
     "src/mcp/server/fastmcp/tools/base.py",
 ]
 
-[tool.pytest.ini_options]
-markers = ["anyio"]
-
 [tool.ruff.lint]
 select = ["E", "F", "I"]
 ignore = []
@@ -93,6 +91,7 @@ members = ["examples/servers/*"]
 mcp = { workspace = true }
 
 [tool.pytest.ini_options]
+markers = ["anyio"]
 xfail_strict = true
 filterwarnings = [
     "error",

src/mcp/server/auth/handlers/authorize.py

@@ -36,8 +36,8 @@ class AuthorizationRequest(BaseModel):
     state: Optional[str] = Field(None, description="Optional state parameter")
     scope: Optional[str] = Field(
         None,
-        description="Optional scope; if specified, should be " \
-            "a space-separated list of scope strings",
+        description="Optional scope; if specified, should be "
+        "a space-separated list of scope strings",
     )
 
     class Config:

src/mcp/server/auth/handlers/revoke.py

@@ -45,9 +45,8 @@ async def revocation_handler(request: Request) -> Response:
         Handler for the OAuth 2.0 Token Revocation endpoint.
         """
         try:
-            revocation_request = RevocationRequest.model_validate_json(
-                await request.body()
-            )
+            form_data = await request.form()
+            revocation_request = RevocationRequest.model_validate(dict(form_data))
         except ValidationError as e:
             raise InvalidRequestError(f"Invalid request body: {e}")
 

src/mcp/server/auth/handlers/token.py

@@ -49,7 +49,8 @@ def create_token_handler(
 ) -> Callable:
     async def token_handler(request: Request):
         try:
-            token_request = TokenRequest.model_validate_json(await request.body()).root
+            form_data = await request.form()
+            token_request = TokenRequest.model_validate(dict(form_data)).root
         except ValidationError as e:
             raise InvalidRequestError(f"Invalid request body: {e}")
         client_info = await client_authenticator(token_request)

src/mcp/server/auth/middleware/client_auth.py

@@ -22,7 +22,7 @@ class ClientAuthRequest(BaseModel):
     """
     Model for client authentication request body.
 
-    Corresponds to ClientAuthenticatedRequestSchema in 
+    Corresponds to ClientAuthenticatedRequestSchema in
     src/server/auth/middleware/clientAuth.ts
     """
 
@@ -32,15 +32,16 @@ class ClientAuthRequest(BaseModel):
 
 class ClientAuthenticator:
     """
-    ClientAuthenticator is a callable which validates requests from a client 
+    ClientAuthenticator is a callable which validates requests from a client
     application,
     used to verify /token and /revoke calls.
-    If, during registration, the client requested to be issued a secret, the 
-    authenticator asserts that /token and /register calls must be authenticated with 
+    If, during registration, the client requested to be issued a secret, the
+    authenticator asserts that /token and /register calls must be authenticated with
     that same token.
-    NOTE: clients can opt for no authentication during registration, in which case this 
+    NOTE: clients can opt for no authentication during registration, in which case this
     logic is skipped.
     """
+
     def __init__(self, clients_store: OAuthRegisteredClientsStore):
         """
         Initialize the dependency.

src/mcp/server/auth/provider.py

@@ -81,10 +81,10 @@ async def create_authorization_code(
         self, client: OAuthClientInformationFull, params: AuthorizationParams
     ) -> str:
         """
-        Generates and stores an authorization code as part of completing the /authorize 
+        Generates and stores an authorization code as part of completing the /authorize
         OAuth step.
 
-        Implementations SHOULD generate an authorization code with at least 160 bits of 
+        Implementations SHOULD generate an authorization code with at least 160 bits of
         entropy,
         and MUST generate an authorization code with at least 128 bits of entropy.
         See https://datatracker.ietf.org/doc/html/rfc6749#section-10.10.

src/mcp/server/sse.py

@@ -132,7 +132,7 @@ async def sse_writer():
             logger.debug("Yielding read and write streams")
             # TODO: hold on; shouldn't we be returning the EventSourceResponse?
             # I think this is why the tests hang
-            # TODO: we probably shouldn't return response here, since it's a breaking 
+            # TODO: we probably shouldn't return response here, since it's a breaking
             # change
             # this is just to test
             yield (read_stream, write_stream, response)

src/mcp/shared/auth.py

@@ -43,19 +43,21 @@ class OAuthClientMetadata(BaseModel):
     """
 
     redirect_uris: List[AnyHttpUrl] = Field(..., min_length=1)
-    # token_endpoint_auth_method: this implementation only supports none & 
+    # token_endpoint_auth_method: this implementation only supports none &
     # client_secret_basic;
     # ie: we do not support client_secret_post
-    token_endpoint_auth_method: Literal["none", "client_secret_basic"] = \
+    token_endpoint_auth_method: Literal["none", "client_secret_basic"] = (
         "client_secret_basic"
+    )
     # grant_types: this implementation only supports authorization_code & refresh_token
-    grant_types: List[Literal["authorization_code", "refresh_token"]] = \
-        ["authorization_code"]
+    grant_types: List[Literal["authorization_code", "refresh_token"]] = [
+        "authorization_code"
+    ]
     # this implementation only supports code; ie: it does not support implicit grants
     response_types: List[Literal["code"]] = ["code"]
     scope: Optional[str] = None
 
-    # these fields are currently unused, but we support & store them for potential 
+    # these fields are currently unused, but we support & store them for potential
     # future use
     client_name: Optional[str] = None
     client_uri: Optional[AnyHttpUrl] = None

tests/server/fastmcp/auth/test_auth_integration.py

@@ -327,6 +327,55 @@ async def test_client_registration(
         # assert await mock_oauth_provider.clients_store.get_client(
         #     client_info["client_id"]
         # ) is not None
+        
+    @pytest.mark.anyio
+    async def test_authorize_form_post(
+        self, test_client: httpx.AsyncClient, mock_oauth_provider: MockOAuthProvider
+    ):
+        """Test the authorization endpoint using POST with form-encoded data."""
+        # Register a client
+        client_metadata = {
+            "redirect_uris": ["https://client.example.com/callback"],
+            "client_name": "Test Client",
+            "grant_types": ["authorization_code", "refresh_token"],
+        }
+
+        response = await test_client.post(
+            "/register",
+            json=client_metadata,
+        )
+        assert response.status_code == 201
+        client_info = response.json()
+
+        # Create a PKCE challenge
+        code_verifier = "some_random_verifier_string"
+        code_challenge = (
+            base64.urlsafe_b64encode(hashlib.sha256(code_verifier.encode()).digest())
+            .decode()
+            .rstrip("=")
+        )
+
+        # Use POST with form-encoded data for authorization
+        response = await test_client.post(
+            "/authorize",
+            data={
+                "response_type": "code",
+                "client_id": client_info["client_id"],
+                "redirect_uri": "https://client.example.com/callback",
+                "code_challenge": code_challenge,
+                "code_challenge_method": "S256",
+                "state": "test_form_state",
+            },
+        )
+        assert response.status_code == 302
+
+        # Extract the authorization code from the redirect URL
+        redirect_url = response.headers["location"]
+        parsed_url = urlparse(redirect_url)
+        query_params = parse_qs(parsed_url.query)
+
+        assert "code" in query_params
+        assert query_params["state"][0] == "test_form_state"
 
     @pytest.mark.anyio
     async def test_authorization_flow(
@@ -337,7 +386,7 @@ async def test_authorization_flow(
         client_metadata = {
             "redirect_uris": ["https://client.example.com/callback"],
             "client_name": "Test Client",
-            "grant_types": ["authorization_code", "refresh_token"]
+            "grant_types": ["authorization_code", "refresh_token"],
         }
 
         response = await test_client.post(
@@ -355,7 +404,7 @@ async def test_authorization_flow(
             .rstrip("=")
         )
 
-        # 3. Request authorization
+        # 3. Request authorization using GET with query params
         response = await test_client.get(
             "/authorize",
             params={
@@ -381,7 +430,7 @@ async def test_authorization_flow(
         # 5. Exchange the authorization code for tokens
         response = await test_client.post(
             "/token",
-            json={
+            data={
                 "grant_type": "authorization_code",
                 "client_id": client_info["client_id"],
                 "client_secret": client_info["client_secret"],
@@ -411,7 +460,7 @@ async def test_authorization_flow(
         # 7. Refresh the token
         response = await test_client.post(
             "/token",
-            json={
+            data={
                 "grant_type": "refresh_token",
                 "client_id": client_info["client_id"],
                 "client_secret": client_info["client_secret"],
@@ -429,7 +478,7 @@ async def test_authorization_flow(
         # 8. Revoke the token
         response = await test_client.post(
             "/revoke",
-            json={
+            data={
                 "client_id": client_info["client_id"],
                 "client_secret": client_info["client_secret"],
                 "token": new_token_response["access_token"],
@@ -505,10 +554,10 @@ def test_tool(x: int) -> str:
             .rstrip("=")
         )
 
-        # Request authorization
-        response = await test_client.get(
+        # Request authorization using POST with form-encoded data
+        response = await test_client.post(
             "/authorize",
-            params={
+            data={
                 "response_type": "code",
                 "client_id": client_info["client_id"],
                 "redirect_uri": "https://client.example.com/callback",
@@ -530,7 +579,7 @@ def test_tool(x: int) -> str:
         # Exchange the authorization code for tokens
         response = await test_client.post(
             "/token",
-            json={
+            data={
                 "grant_type": "authorization_code",
                 "client_id": client_info["client_id"],
                 "client_secret": client_info["client_secret"],