Merge branch 'v1.x' into fix/v1x-nonutf8-post-body-parse-error

Author: tirthfx

docs/server.md

@@ -1253,7 +1253,7 @@ The FastMCP server instance accessible via `ctx.fastmcp` provides access to serv
   - `host` and `port` - Server network configuration
   - `mount_path`, `sse_path`, `streamable_http_path` - Transport paths
   - `stateless_http` - Whether the server operates in stateless mode
-  - `max_request_body_size` - Maximum Streamable HTTP POST body size in bytes
+  - `max_request_body_size` - Maximum HTTP request body size in bytes (Streamable HTTP and SSE)
   - And other configuration options
 
 ```python
@@ -1418,9 +1418,9 @@ Note that `uv run mcp run` or `uv run mcp dev` only supports server using FastMC
 
 > **Note**: Streamable HTTP transport is the recommended transport for production deployments. Use `stateless_http=True` and `json_response=True` for optimal scalability.
 
-Streamable HTTP POST bodies are limited to 4 MiB by default. Larger requests receive HTTP 413
-before parsing or session creation. If your server intentionally accepts larger MCP messages,
-configure the smallest suitable byte limit:
+HTTP request bodies (Streamable HTTP and SSE) are limited to 4 MiB by default. Larger requests
+receive HTTP 413 before parsing or session creation. If your server intentionally accepts larger MCP
+messages, configure the smallest suitable byte limit:
 
 ```python
 mcp = FastMCP("Large messages", max_request_body_size=8 * 1024 * 1024)

src/mcp/client/session.py

@@ -428,17 +428,24 @@ async def _validate_tool_result(self, name: str, result: types.CallToolResult) -
 
         if output_schema is not None:
             from jsonschema import SchemaError, ValidationError, validate
+            from referencing import Registry
+            from referencing.exceptions import Unresolvable
 
             if result.structuredContent is None:
                 raise RuntimeError(
                     f"Tool {name} has an output schema but did not return structured content"
                 )  # pragma: no cover
+            # An explicit empty registry: `$ref`s resolve within the schema document and the bundled metaschemas.
+            registry: Registry[Any] = Registry()
             try:
-                validate(result.structuredContent, output_schema)
+                validate(result.structuredContent, output_schema, registry=registry)
             except ValidationError as e:
                 raise RuntimeError(f"Invalid structured content returned by tool {name}: {e}")  # pragma: no cover
             except SchemaError as e:  # pragma: no cover
                 raise RuntimeError(f"Invalid schema for tool {name}: {e}")  # pragma: no cover
+            except Unresolvable as e:
+                # A `$ref` did not resolve within the schema document.
+                raise RuntimeError(f"Invalid schema for tool {name}: {e}") from e
 
     @overload
     @deprecated("Use list_prompts(params=PaginatedRequestParams(...)) instead")

src/mcp/server/auth/routes.py

@@ -18,6 +18,7 @@
 from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
 from mcp.server.streamable_http import MCP_PROTOCOL_VERSION_HEADER
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, RequestBodyLimitMiddleware
 from mcp.shared.auth import OAuthMetadata
 
 
@@ -53,17 +54,24 @@ def validate_issuer_url(url: AnyHttpUrl):
 REVOCATION_PATH = "/revoke"
 
 
-def cors_middleware(
-    handler: Callable[[Request], Response | Awaitable[Response]],
-    allow_methods: list[str],
-) -> ASGIApp:
-    cors_app = CORSMiddleware(
-        app=request_response(handler),
+def _cors(app: ASGIApp, allow_methods: list[str]) -> ASGIApp:
+    return CORSMiddleware(
+        app=app,
         allow_origins="*",
         allow_methods=allow_methods,
         allow_headers=[MCP_PROTOCOL_VERSION_HEADER],
     )
-    return cors_app
+
+
+def _body_limited(app: ASGIApp) -> ASGIApp:
+    return RequestBodyLimitMiddleware(app, DEFAULT_MAX_REQUEST_BODY_SIZE)
+
+
+def cors_middleware(
+    handler: Callable[[Request], Response | Awaitable[Response]],
+    allow_methods: list[str],
+) -> ASGIApp:
+    return _cors(request_response(handler), allow_methods)
 
 
 def create_auth_routes(
@@ -84,11 +92,13 @@ def create_auth_routes(
         revocation_options,
     )
     client_authenticator = ClientAuthenticator(provider)
+    token_handler = TokenHandler(provider, client_authenticator)
 
     # Create routes
     # Allow CORS requests for endpoints meant to be hit by the OAuth client
     # (with the client secret). This is intended to support things like MCP Inspector,
-    # where the client runs in a web browser.
+    # where the client runs in a web browser. CORS is the outermost wrapper so that
+    # responses produced by inner layers (such as a 413) still carry CORS headers.
     routes = [
         Route(
             "/.well-known/oauth-authorization-server",
@@ -102,15 +112,12 @@ def create_auth_routes(
             AUTHORIZATION_PATH,
             # do not allow CORS for authorization endpoint;
             # clients should just redirect to this
-            endpoint=AuthorizationHandler(provider).handle,
+            endpoint=_body_limited(request_response(AuthorizationHandler(provider).handle)),
             methods=["GET", "POST"],
         ),
         Route(
             TOKEN_PATH,
-            endpoint=cors_middleware(
-                TokenHandler(provider, client_authenticator).handle,
-                ["POST", "OPTIONS"],
-            ),
+            endpoint=_cors(_body_limited(request_response(token_handler.handle)), ["POST", "OPTIONS"]),
             methods=["POST", "OPTIONS"],
         ),
     ]
@@ -123,10 +130,7 @@ def create_auth_routes(
         routes.append(
             Route(
                 REGISTRATION_PATH,
-                endpoint=cors_middleware(
-                    registration_handler.handle,
-                    ["POST", "OPTIONS"],
-                ),
+                endpoint=_cors(_body_limited(request_response(registration_handler.handle)), ["POST", "OPTIONS"]),
                 methods=["POST", "OPTIONS"],
             )
         )
@@ -136,10 +140,7 @@ def create_auth_routes(
         routes.append(
             Route(
                 REVOCATION_PATH,
-                endpoint=cors_middleware(
-                    revocation_handler.handle,
-                    ["POST", "OPTIONS"],
-                ),
+                endpoint=_cors(_body_limited(request_response(revocation_handler.handle)), ["POST", "OPTIONS"]),
                 methods=["POST", "OPTIONS"],
             )
         )

src/mcp/server/fastmcp/server.py

@@ -62,8 +62,8 @@
 from mcp.server.sse import SseServerTransport
 from mcp.server.stdio import stdio_server
 from mcp.server.streamable_http import EventStore
-from mcp.server.streamable_http_manager import DEFAULT_MAX_REQUEST_BODY_SIZE, StreamableHTTPSessionManager
-from mcp.server.transport_security import TransportSecuritySettings
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import DEFAULT_MAX_REQUEST_BODY_SIZE, TransportSecuritySettings
 from mcp.shared.context import LifespanContextT, RequestContext, RequestT
 from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
 from mcp.types import Prompt as MCPPrompt
@@ -107,6 +107,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
     stateless_http: bool
     """Define if the server should create a new transport per request."""
     max_request_body_size: int
+    """Maximum request body size in bytes for the Streamable HTTP endpoint and the SSE message endpoint."""
 
     # resource settings
     warn_on_duplicate_resources: bool
@@ -835,6 +836,7 @@ def sse_app(self, mount_path: str | None = None) -> Starlette:
         sse = SseServerTransport(
             normalized_message_endpoint,
             security_settings=self.settings.transport_security,
+            max_request_body_size=self.settings.max_request_body_size,
         )
 
         async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no cover
@@ -1087,6 +1089,11 @@ async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -
             raise ValueError(str(e))
 
 
+# `Settings.lifespan` refers to FastMCP, which is only defined above; complete the model now so
+# settings sources never see an unresolved annotation when a FastMCP instance is created.
+Settings.model_rebuild()
+
+
 class StreamableHTTPASGIApp:
     """
     ASGI application for Streamable HTTP server transport.

src/mcp/server/fastmcp/utilities/func_metadata.py

@@ -45,6 +45,25 @@ def emit_warning(self, kind: JsonSchemaWarningKind, detail: str) -> None:
         raise ValueError(f"JSON schema warning: {kind} - {detail}")
 
 
+_LOCAL_DEFS_PREFIX = "#/$defs/"
+
+
+def _inline_root_ref(schema: dict[str, Any]) -> dict[str, Any]:
+    """Give a schema whose root is a bare `$ref` into `$defs` an inline root.
+
+    pydantic emits a self-referential model as `{"$defs": {...}, "$ref": "#/$defs/Model"}`, with no
+    `type` at the root; `Tool.outputSchema` requires `type: object` at the root. The referenced
+    definition is copied onto the root and `$defs` is kept, since nested references still point into
+    it. Root siblings of the `$ref` win over the definition's keys.
+    """
+    ref = schema.get("$ref")
+    if not isinstance(ref, str) or not ref.startswith(_LOCAL_DEFS_PREFIX):
+        return schema
+    definition = cast(dict[str, Any], schema["$defs"][ref.removeprefix(_LOCAL_DEFS_PREFIX)])
+    siblings = {key: value for key, value in schema.items() if key != "$ref"}
+    return {**definition, **siblings}
+
+
 class ArgModelBase(BaseModel):
     """A model representing the arguments to a function."""
 
@@ -428,7 +447,7 @@ def _try_create_model_and_schema(
             logger.info(f"Cannot create schema for type {type_expr} in {func_name}: {type(e).__name__}: {e}")
             return None, None, False
 
-        return model, schema, wrap_output
+        return model, _inline_root_ref(schema), wrap_output
 
     return None, None, False
 

src/mcp/server/sse.py

@@ -54,6 +54,8 @@ async def handle_sse(request):
 import mcp.types as types
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser, AuthorizationContext, authorization_context
 from mcp.server.transport_security import (
+    DEFAULT_MAX_REQUEST_BODY_SIZE,
+    RequestBodyLimitMiddleware,
     TransportSecurityMiddleware,
     TransportSecuritySettings,
 )
@@ -81,7 +83,12 @@ class SseServerTransport:
     _session_owners: dict[UUID, AuthorizationContext]
     _security: TransportSecurityMiddleware
 
-    def __init__(self, endpoint: str, security_settings: TransportSecuritySettings | None = None) -> None:
+    def __init__(
+        self,
+        endpoint: str,
+        security_settings: TransportSecuritySettings | None = None,
+        max_request_body_size: int = DEFAULT_MAX_REQUEST_BODY_SIZE,
+    ) -> None:
         """
         Creates a new SSE server transport, which will direct the client to POST
         messages to the relative path given.
@@ -90,6 +97,9 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
             endpoint: A relative path where messages should be posted
                     (e.g., "/messages/").
             security_settings: Optional security settings for DNS rebinding protection.
+            max_request_body_size: Maximum size in bytes for POSTed message bodies. Requests that
+                declare or stream a larger body receive HTTP 413. Defaults to 4 MiB, matching
+                `StreamableHTTPSessionManager`.
 
         Note:
             We use relative paths instead of full URLs for several reasons:
@@ -106,6 +116,9 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
 
         super().__init__()
 
+        if max_request_body_size <= 0:
+            raise ValueError("max_request_body_size must be a positive number of bytes")
+
         # Validate that endpoint is a relative path and not a full URL
         if "://" in endpoint or endpoint.startswith("//") or "?" in endpoint or "#" in endpoint:
             raise ValueError(
@@ -121,6 +134,7 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
         self._read_stream_writers = {}
         self._session_owners = {}
         self._security = TransportSecurityMiddleware(security_settings)
+        self._post_message_app = RequestBodyLimitMiddleware(self._handle_post_message, max_request_body_size)
         logger.debug(f"SseServerTransport initialized with endpoint: {endpoint}")
 
     @asynccontextmanager
@@ -214,7 +228,18 @@ async def response_wrapper(scope: Scope, receive: Receive, send: Send):
             self._read_stream_writers.pop(session_id, None)
             self._session_owners.pop(session_id, None)
 
-    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
+    async def handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:
+        """ASGI application for the message endpoint.
+
+        Only POST is accepted (other methods get 405), and bodies larger than
+        `max_request_body_size` are answered with 413 before the message is handled.
+        """
+        if scope["method"] != "POST":
+            response = Response(status_code=405, headers={"Allow": "POST"})
+            return await response(scope, receive, send)
+        await self._post_message_app(scope, receive, send)
+
+    async def _handle_post_message(self, scope: Scope, receive: Receive, send: Send) -> None:  # pragma: no cover
         logger.debug("Handling POST message")
         request = Request(scope, receive)