Review feedback

Author: praboud-ant

src/mcp/server/auth/errors.py

@@ -1,31 +1,4 @@
-from typing import Literal
-
-from pydantic import BaseModel, ValidationError
-
-ErrorCode = Literal["invalid_request", "invalid_client"]
-
-
-class ErrorResponse(BaseModel):
-    error: ErrorCode
-    error_description: str
-
-
-class OAuthError(Exception):
-    """
-    Base class for all OAuth errors.
-    """
-
-    error_code: ErrorCode
-
-    def __init__(self, error_description: str):
-        super().__init__(error_description)
-        self.error_description = error_description
-
-    def error_response(self) -> ErrorResponse:
-        return ErrorResponse(
-            error=self.error_code,
-            error_description=self.error_description,
-        )
+from pydantic import ValidationError
 
 
 def stringify_pydantic_error(validation_error: ValidationError) -> str:

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

@@ -1,15 +1,13 @@
 import logging
 from dataclasses import dataclass
 from typing import Any, Literal
-from urllib.parse import urlencode, urlparse, urlunparse
 
 from pydantic import AnyHttpUrl, AnyUrl, BaseModel, Field, RootModel, ValidationError
 from starlette.datastructures import FormData, QueryParams
 from starlette.requests import Request
 from starlette.responses import RedirectResponse, Response
 
 from mcp.server.auth.errors import (
-    OAuthError,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
@@ -225,34 +223,3 @@ async def error_response(
             return await error_response(
                 error="server_error", error_description="An unexpected error occurred"
             )
-
-
-def create_error_redirect(
-    redirect_uri: AnyUrl, error: Exception | AuthorizationErrorResponse
-) -> str:
-    parsed_uri = urlparse(str(redirect_uri))
-
-    if isinstance(error, AuthorizationErrorResponse):
-        # Convert ErrorResponse to dict
-        error_dict = error.model_dump(exclude_none=True)
-        query_params = {}
-        for key, value in error_dict.items():
-            if value is not None:
-                if key == "error_uri" and hasattr(value, "__str__"):
-                    query_params[key] = str(value)
-                else:
-                    query_params[key] = value
-
-    elif isinstance(error, OAuthError):
-        query_params = {"error": error.error_code, "error_description": str(error)}
-    else:
-        query_params = {
-            "error": "server_error",
-            "error_description": "An unknown error occurred",
-        }
-
-    new_query = urlencode(query_params)
-    if parsed_uri.query:
-        new_query = f"{parsed_uri.query}&{new_query}"
-
-    return urlunparse(parsed_uri._replace(query=new_query))

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

@@ -8,7 +8,6 @@
 from starlette.requests import Request
 
 from mcp.server.auth.errors import (
-    ErrorResponse,
     stringify_pydantic_error,
 )
 from mcp.server.auth.json_response import PydanticJSONResponse
@@ -80,7 +79,7 @@ class TokenHandler:
     provider: OAuthServerProvider[Any, Any, Any]
     client_authenticator: ClientAuthenticator
 
-    def response(self, obj: TokenSuccessResponse | TokenErrorResponse | ErrorResponse):
+    def response(self, obj: TokenSuccessResponse | TokenErrorResponse):
         status_code = 200
         if isinstance(obj, TokenErrorResponse):
             status_code = 400

src/mcp/server/auth/provider.py

@@ -154,7 +154,7 @@ async def authorize(
         +------------+
 
         Implementations will need to define another handler on the MCP server return
-        flow to perform the second redirect, and generates and stores an authorization
+        flow to perform the second redirect, and generate and store an authorization
         code as part of completing the OAuth authorization step.
 
         Implementations SHOULD generate an authorization code with at least 160 bits of

src/mcp/server/fastmcp/server.py

@@ -489,6 +489,27 @@ def custom_route(
         name: str | None = None,
         include_in_schema: bool = True,
     ):
+        """
+        Decorator to register a custom HTTP route on the FastMCP server.
+
+        Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
+        which can be useful for OAuth callbacks, health checks, or admin APIs.
+        The handler function must be an async function that accepts a Starlette
+        Request and returns a Response.
+
+        Args:
+            path: URL path for the route (e.g., "/oauth/callback")
+            methods: List of HTTP methods to support (e.g., ["GET", "POST"])
+            name: Optional name for the route (to reference this route with
+                  Starlette's reverse URL lookup feature)
+            include_in_schema: Whether to include in OpenAPI schema, defaults to True
+
+        Example:
+            @server.custom_route("/health", methods=["GET"])
+            async def health_check(request: Request) -> Response:
+                return JSONResponse({"status": "ok"})
+        """
+
         def decorator(
             func: Callable[[Request], Awaitable[Response]],
         ) -> Callable[[Request], Awaitable[Response]]:
@@ -517,6 +538,7 @@ async def run_stdio_async(self) -> None:
     async def run_sse_async(self) -> None:
         """Run the server using SSE transport."""
         import uvicorn
+
         starlette_app = self.sse_app()
 
         config = uvicorn.Config(
@@ -529,6 +551,7 @@ async def run_sse_async(self) -> None:
         await server.serve()
 
     def sse_app(self) -> Starlette:
+        """Return an instance of the SSE server app."""
         from starlette.middleware import Middleware
         from starlette.routing import Mount, Route
 

tests/server/fastmcp/servers/test_file_server.py

@@ -114,17 +114,13 @@ async def test_read_resource_file(mcp: FastMCP):
 
 @pytest.mark.anyio
 async def test_delete_file(mcp: FastMCP, test_dir: Path):
-    await mcp.call_tool(
-        "delete_file", arguments={"path": str(test_dir / "example.py")}
-    )
+    await mcp.call_tool("delete_file", arguments={"path": str(test_dir / "example.py")})
     assert not (test_dir / "example.py").exists()
 
 
 @pytest.mark.anyio
 async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
-    await mcp.call_tool(
-        "delete_file", arguments={"path": str(test_dir / "example.py")}
-    )
+    await mcp.call_tool("delete_file", arguments={"path": str(test_dir / "example.py")})
     res_iter = await mcp.read_resource("file://test_dir/example.py")
     res_list = list(res_iter)
     assert len(res_list) == 1