Forward proxy server metadata across protocol eras (#4776)

* Forward proxy negotiation metadata

🤖 Generated with OpenAI Codex

* Limit forwarded proxy metadata

🤖 Generated with OpenAI Codex

* Tighten negotiation metadata forwarding

🤖 Generated with OpenAI Codex

* Tighten proxy metadata docs

🤖 Generated with OpenAI Codex

* Keep proxy metadata middleware with provider

🤖 Generated with OpenAI Codex

* Simplify proxy negotiation middleware API

🤖 Generated with OpenAI Codex

* Name proxy metadata middleware directly

🤖 Generated with OpenAI Codex

* Preserve discovery middleware contracts

🤖 Generated with OpenAI Codex

* Clarify proxy metadata ownership

🤖 Generated with OpenAI Codex

* Align proxy metadata wording

🤖 Generated with OpenAI Codex

* Call forwarded values server metadata

🤖 Generated with OpenAI Codex

* Harden proxy metadata reads

🤖 Generated with OpenAI Codex

* Expose configured discovery result

🤖 Generated with OpenAI Codex

* Preserve proxy discovery compatibility

🤖 Generated with OpenAI Codex

* Preserve deprecated initialization middleware

🤖 Generated with OpenAI Codex

* Harden proxy metadata boundaries

🤖 Generated with OpenAI Codex

* Restore deprecated middleware location

🤖 Generated with OpenAI Codex

* Simplify proxy metadata client lifecycle

🤖 Generated with OpenAI Codex

* Clarify proxy metadata lifecycle

🤖 Generated with OpenAI Codex

* Preserve proxy factory errors

🤖 Generated with OpenAI Codex

* Detach forwarded proxy metadata

🤖 Generated with OpenAI Codex

* Simplify proxy metadata implementation

🤖 Generated with OpenAI Codex

* Distinguish proxy metadata failures

🤖 Generated with OpenAI Codex

* Narrow proxy metadata validation fallback

🤖 Generated with OpenAI Codex

* Retrigger CI

🤖 Generated with OpenAI Codex

Author: jakekaplan

docs/servers/middleware.mdx

@@ -310,6 +310,22 @@ async def on_initialize(self, context: MiddlewareContext, call_next):
 Rejection works only **before** `call_next()`. Raising `McpError` afterward logs the error without sending it — the client still receives a successful initialize response.
 </Warning>
 
+#### on_discover
+
+Called when a modern client negotiates through `server/discover`. Core discovery responses are returned as `DiscoverResult`; extension-owned result types are returned as dictionaries and should be passed through unless the middleware handles that extension.
+
+```python
+from mcp_types import DiscoverResult
+
+async def on_discover(self, context, call_next):
+    result = await call_next(context)
+    if not isinstance(result, DiscoverResult):
+        return result
+    return result.model_copy(update={"instructions": "Custom instructions"})
+```
+
+Fields such as `supported_versions`, `capabilities`, and cache policy should only be changed when the server's public behavior also changes.
+
 ### Raw Handler
 
 For complete control over all messages, override `__call__` instead of individual hooks:

docs/servers/providers/proxy.mdx

@@ -60,11 +60,9 @@ To mount a proxy inside another FastMCP server, see [Mounting External Servers](
 
 ## Connection Semantics
 
-FastMCP proxies are lazy bridges. Creating the proxy object and starting the local server do not contact the upstream server. The upstream connection begins when an MCP client sends an `initialize` request to the proxy.
+FastMCP proxies are lazy bridges. Creating the proxy object and starting the local server do not contact the upstream server. During client negotiation, the proxy makes a best-effort request for optional server metadata using the backend client's existing lifecycle and negotiation mode; an unavailable backend does not prevent the client from connecting to the proxy.
 
-During initialization, the proxy initializes the upstream server before responding locally. If the upstream server is unavailable, the URL does not point to an MCP endpoint, or upstream authentication cannot complete, the proxy initialization fails. This keeps the local proxy's connection status aligned with the upstream server it represents.
-
-After initialization, the proxy forwards MCP requests such as `ping`, `tools/list`, `resources/list`, `prompts/list`, tool calls, resource reads, sampling, elicitation, logging, and progress through the upstream client.
+Subsequent MCP requests such as `ping`, `tools/list`, `resources/list`, `prompts/list`, tool calls, resource reads, sampling, elicitation, logging, and progress connect to the backend as needed. Component provider failures follow `provider_error_strategy`: the default `"warn"` logs and skips a failed provider, while `"raise"` reports the failure to the client.
 
 ## Transport Bridging
 
@@ -388,6 +386,28 @@ Only reuse sessions when you know the backend is stateless (e.g. stateless HTTP)
 
 ## Advanced Usage
 
+### Forwarding Server Metadata
+
+Add `ProxyMetadataMiddleware` when a gateway built with `ProxyProvider` should also expose backend instructions and namespaced `_meta`:
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.providers.proxy import (
+    ProxyClient,
+    ProxyMetadataMiddleware,
+    ProxyProvider,
+)
+
+backend = ProxyProvider(lambda: ProxyClient("http://backend:8000/mcp", mode="auto"))
+gateway = FastMCP(
+    "Controlled Gateway",
+    providers=[backend],
+    middleware=[ProxyMetadataMiddleware(backend)],
+)
+```
+
+By default the gateway keeps its own `serverInfo`; pass `identity="upstream"` to use the backend identity when available. Frontend instructions and `_meta` values win on collisions. The middleware never copies upstream protocol versions, connection metadata, capabilities, cache policy, `resultType`, or unknown top-level fields. If the backend is unavailable, the client can still connect without its optional metadata.
+
 ### FastMCPProxy Class
 
 For explicit session control, use `FastMCPProxy` directly:

fastmcp_slim/fastmcp/client/client.py

@@ -657,6 +657,11 @@ def session(self) -> ClientSession:
 
         return self._session_state.session
 
+    @property
+    def prior_discover(self) -> mcp_types.DiscoverResult | None:
+        """The configured result to adopt when `mode` pins a modern version."""
+        return self._prior_discover
+
     @property
     def initialize_result(self) -> mcp_types.InitializeResult | None:
         """Get the result of the initialization request.

fastmcp_slim/fastmcp/server/low_level.py

@@ -153,10 +153,11 @@ class FastMCPServerMiddleware:
 
     Dispatch shapes:
 
-    - ``initialize`` runs the *whole* FastMCP chain here (``on_message`` ->
-      ``on_request`` -> ``on_initialize``) because there is no interior handler
-      adapter for it: the SDK builds the ``InitializeResult`` directly, so this is
-      the only place ``on_initialize`` can observe it or veto with ``MCPError``.
+    - Negotiation runs the *whole* FastMCP chain here: ``initialize`` dispatches
+      through ``on_initialize`` and ``server/discover`` through ``on_discover``.
+      Neither has an interior FastMCP handler adapter, and the SDK serializes both
+      results before returning through its middleware seam, so this root adapter
+      restores core results to typed models before FastMCP middleware observes them.
     - The component methods (``tools/call``, ``tools/list``, ``resources/read``,
       ...) still run their FastMCP chain *interior*, in the handler adapter, where
       ``on_call_tool`` receives the typed component result and a tool exception
@@ -192,6 +193,8 @@ async def __call__(
                 return await call_next(ctx)
             if ctx.method == "initialize" and ctx.request_id is not None:
                 return await self._run_initialize_mw(fastmcp, ctx, call_next)
+            if ctx.method == "server/discover" and ctx.request_id is not None:
+                return await self._run_discover_mw(fastmcp, ctx, call_next)
             if ctx.request_id is not None and ctx.method in _INTERIOR_METHODS:
                 return await self._dispatch_component(fastmcp, ctx, call_next)
             return await self._run_outer_mw(fastmcp, ctx, call_next, _raise=None)
@@ -318,6 +321,62 @@ def _apply_shared_context(self, fastmcp: FastMCP | None) -> Iterator[None]:
             for var, token in reversed(tokens):
                 var.reset(token)
 
+    async def _run_discover_mw(
+        self,
+        fastmcp: FastMCP,
+        ctx: ServerRequestContext,
+        call_next: CallNext,
+    ) -> HandlerResult:
+        """Run discovery through the typed FastMCP middleware hook."""
+        from fastmcp.server.context import Context
+        from fastmcp.server.middleware.middleware import MiddlewareContext
+
+        try:
+            discover_message = mcp_types.DiscoverRequest.model_validate(
+                {"method": "server/discover", "params": ctx.params}, by_name=False
+            )
+        except ValidationError as exc:
+            return await self._run_outer_mw(fastmcp, ctx, call_next, _raise=exc)
+
+        async def call_original_handler(
+            _mw_ctx: MiddlewareContext,
+        ) -> mcp_types.DiscoverResult | dict[str, Any]:
+            message = _mw_ctx.message
+            params = (
+                message.params.model_dump(by_alias=True, mode="json", exclude_none=True)
+                if message.params is not None
+                else None
+            )
+            raw = await call_next(replace(ctx, params=params))
+            if isinstance(raw, mcp_types.DiscoverResult):
+                return raw
+            if isinstance(raw, Mapping):
+                result = dict(raw)
+                result_type = result.get("resultType")
+                if (
+                    isinstance(result_type, str)
+                    and result_type not in mcp_types.CORE_RESULT_TYPES
+                ):
+                    return result
+                return mcp_types.DiscoverResult.model_validate(result)
+            raise TypeError(
+                "server/discover handler returned "
+                f"{type(raw).__name__}; expected DiscoverResult or mapping"
+            )
+
+        async with Context(fastmcp=fastmcp, session=ctx.session) as fastmcp_ctx:
+            mw_context = MiddlewareContext(
+                message=discover_message,
+                source="client",
+                type="request",
+                method="server/discover",
+                fastmcp_context=fastmcp_ctx,
+            )
+            return await fastmcp._run_middleware(
+                mw_context,
+                cast("FastMCPCallNext[Any, Any]", call_original_handler),
+            )
+
     async def _run_initialize_mw(
         self,
         fastmcp: FastMCP,

fastmcp_slim/fastmcp/server/middleware/middleware.py

@@ -170,6 +170,8 @@ async def _dispatch_handler(
             match context.method:
                 case "initialize":
                     handler = make_handler_wrapper(self.on_initialize, handler)
+                case "server/discover":
+                    handler = make_handler_wrapper(self.on_discover, handler)
                 case "tools/call":
                     handler = make_handler_wrapper(self.on_call_tool, handler)
                 case "resources/read":
@@ -227,6 +229,13 @@ async def on_initialize(
     ) -> mt.InitializeResult | None:
         return await call_next(context)
 
+    async def on_discover(
+        self,
+        context: MiddlewareContext[mt.DiscoverRequest],
+        call_next: CallNext[mt.DiscoverRequest, mt.DiscoverResult | dict[str, Any]],
+    ) -> mt.DiscoverResult | dict[str, Any]:
+        return await call_next(context)
+
     async def on_call_tool(
         self,
         context: MiddlewareContext[mt.CallToolRequestParams],