feat: ServerRunner builds InitializeResult from InitializationOptions

ServerRunner gains init_options (defaults to
server.create_initialization_options()). _handle_initialize builds the full
InitializeResult from it (name/title/description/version/website_url/icons/
instructions) and negotiates requested-if-in-SUPPORTED_PROTOCOL_VERSIONS-
else-LATEST, matching ServerSession._received_request.

Author: maxisbey

src/mcp/server/runner.py

@@ -29,10 +29,12 @@
 from mcp.server.connection import Connection
 from mcp.server.context import CallNext, Context, ServerMiddleware
 from mcp.server.lowlevel.server import Server
+from mcp.server.models import InitializationOptions
 from mcp.shared._otel import extract_trace_context, otel_span
 from mcp.shared.dispatcher import DispatchContext, Dispatcher, DispatchMiddleware, OnRequest
 from mcp.shared.exceptions import MCPError
 from mcp.shared.transport_context import TransportContext
+from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import (
     INVALID_PARAMS,
     LATEST_PROTOCOL_VERSION,
@@ -125,6 +127,8 @@ class ServerRunner(Generic[LifespanT]):
     dispatcher: Dispatcher[TransportContext]
     lifespan_state: LifespanT
     has_standalone_channel: bool
+    init_options: InitializationOptions | None = None
+    """`InitializeResult` payload. Defaults to `server.create_initialization_options()`."""
     session_id: str | None = None
     stateless: bool = False
     dispatch_middleware: list[DispatchMiddleware] = field(default_factory=list[DispatchMiddleware])
@@ -134,6 +138,8 @@ class ServerRunner(Generic[LifespanT]):
 
     def __post_init__(self) -> None:
         self._initialized = self.stateless
+        if self.init_options is None:
+            self.init_options = self.server.create_initialization_options()
         self.connection = Connection(
             self.dispatcher, has_standalone_channel=self.has_standalone_channel, session_id=self.session_id
         )
@@ -234,18 +240,24 @@ def _handle_initialize(self, params: Mapping[str, Any] | None) -> dict[str, Any]
         init = InitializeRequestParams.model_validate(params or {})
         self.connection.client_info = init.client_info
         self.connection.client_capabilities = init.capabilities
-        # TODO: real version negotiation. This always responds with LATEST,
-        # which is wrong - the server should pick the highest version both
-        # sides support and compute a per-connection feature set from it.
-        # See FOLLOWUPS: "Consolidate per-connection mode/negotiation".
-        self.connection.protocol_version = (
-            init.protocol_version if init.protocol_version in {LATEST_PROTOCOL_VERSION} else LATEST_PROTOCOL_VERSION
-        )
+        requested = init.protocol_version
+        negotiated = requested if requested in SUPPORTED_PROTOCOL_VERSIONS else LATEST_PROTOCOL_VERSION
+        self.connection.protocol_version = negotiated
         self._initialized = True
         self.connection.initialized.set()
+        assert self.init_options is not None
+        opts = self.init_options
         result = InitializeResult(
-            protocol_version=self.connection.protocol_version,
-            capabilities=self.server.capabilities(),
-            server_info=Implementation(name=self.server.name, version=self.server.version or "0.0.0"),
+            protocol_version=negotiated,
+            capabilities=opts.capabilities,
+            server_info=Implementation(
+                name=opts.server_name,
+                title=opts.title,
+                description=opts.description,
+                version=opts.server_version,
+                website_url=opts.website_url,
+                icons=opts.icons,
+            ),
+            instructions=opts.instructions,
         )
         return _dump_result(result)

tests/server/test_runner.py

@@ -18,10 +18,12 @@
 from mcp.server.connection import Connection
 from mcp.server.context import Context
 from mcp.server.lowlevel.server import NotificationOptions, Server
+from mcp.server.models import InitializationOptions
 from mcp.server.runner import ServerRunner, otel_middleware
 from mcp.shared.direct_dispatcher import DirectDispatcher, create_direct_dispatcher_pair
 from mcp.shared.dispatcher import DispatchMiddleware
 from mcp.shared.exceptions import MCPError
+from mcp.shared.version import SUPPORTED_PROTOCOL_VERSIONS
 from mcp.types import (
     INTERNAL_ERROR,
     INVALID_PARAMS,
@@ -77,6 +79,7 @@ async def connected_runner(
     initialized: bool = True,
     stateless: bool = False,
     has_standalone_channel: bool = True,
+    init_options: InitializationOptions | None = None,
     session_id: str | None = None,
     headers: Mapping[str, str] | None = None,
     dispatch_middleware: list[DispatchMiddleware] | None = None,
@@ -94,6 +97,7 @@ async def connected_runner(
         dispatcher=server_d,
         lifespan_state={},
         has_standalone_channel=has_standalone_channel,
+        init_options=init_options,
         session_id=session_id,
         stateless=stateless,
         dispatch_middleware=dispatch_middleware or [],
@@ -327,20 +331,31 @@ async def greet(ctx: Any, params: GreetParams) -> dict[str, Any]:
 
 
 @pytest.mark.anyio
-async def test_server_capabilities_reflects_ctor_options_in_initialize_result():
+async def test_runner_initialize_result_reflects_init_options():
     async def list_tools(ctx: Any, params: PaginatedRequestParams | None) -> ListToolsResult:
         raise NotImplementedError
 
-    server: SrvT = Server(
-        name="caps-test",
-        on_list_tools=list_tools,
-        notification_options=NotificationOptions(tools_changed=True),
-        experimental_capabilities={"ext": {"k": "v"}},
-    )
-    async with connected_runner(server, initialized=False) as (client, _):
+    server: SrvT = Server(name="caps-test", on_list_tools=list_tools, instructions="be nice")
+    init_options = server.create_initialization_options(NotificationOptions(tools_changed=True), {"ext": {"k": "v"}})
+    async with connected_runner(server, initialized=False, init_options=init_options) as (client, _):
         result = await client.send_raw_request("initialize", _initialize_params())
     assert result["capabilities"]["tools"]["listChanged"] is True
     assert result["capabilities"]["experimental"] == {"ext": {"k": "v"}}
+    assert result["serverInfo"]["name"] == "caps-test"
+    assert result["instructions"] == "be nice"
+
+
+@pytest.mark.anyio
+async def test_runner_initialize_echoes_supported_version_and_falls_back_to_latest(server: SrvT):
+    oldest = SUPPORTED_PROTOCOL_VERSIONS[0]
+    async with connected_runner(server, initialized=False) as (client, _):
+        params = {**_initialize_params(), "protocolVersion": oldest}
+        result = await client.send_raw_request("initialize", params)
+        assert result["protocolVersion"] == oldest
+    async with connected_runner(server, initialized=False) as (client, _):
+        params = {**_initialize_params(), "protocolVersion": "1999-01-01"}
+        result = await client.send_raw_request("initialize", params)
+        assert result["protocolVersion"] == LATEST_PROTOCOL_VERSION
 
 
 @pytest.mark.anyio