test: rebaseline legacy unit tests for the dispatcher swap; rewrite __main__

server/__main__.py: use Server.run() instead of manual ServerSession +
incoming_messages. 49 -> 24 lines.

tests/server/test_runner.py: connected_runner now drives a JSONRPCDispatcher
pair (was DirectDispatcher); ctx is ServerRequestContext; dropped 6 tests
of dormant Context features that return in the Context rework.

tests/server/test_connection.py: set client_params instead of the now
read-only client_info/client_capabilities properties.

tests/server/test_stateless_mode.py: ServerSession(dispatcher, connection)
fixtures.

tests/conftest.py: capfire override resets _otel._tracer to NoOpTracer on
teardown so tests after a span-capture test don't see traceparent injected
into _meta (pre-existing order-dep, surfaced once both code paths inject).

Deleted (covered by tests/interaction/ or test_runner.py):
test_session.py, test_session_race_condition.py,
test_progress_notifications.py, test_malformed_input.py.

test_lowlevel_exception_handling.py: dropped 3 tests of the deleted
_handle_message; kept the real-stream Server.run() regression test.

Author: maxisbey

src/mcp/server/__main__.py

@@ -1,14 +1,11 @@
-import importlib.metadata
 import logging
 import sys
 import warnings
 
 import anyio
 
-from mcp.server.models import InitializationOptions
-from mcp.server.session import ServerSession
+from mcp.server.lowlevel.server import Server
 from mcp.server.stdio import stdio_server
-from mcp.types import ServerCapabilities
 
 if not sys.warnoptions:
     warnings.simplefilter("ignore")
@@ -17,32 +14,10 @@
 logger = logging.getLogger("server")
 
 
-async def receive_loop(session: ServerSession):
-    logger.info("Starting receive loop")
-    async for message in session.incoming_messages:
-        if isinstance(message, Exception):
-            logger.error("Error: %s", message)
-            continue
-
-        logger.info("Received message from client: %s", message)
-
-
-async def main():
-    version = importlib.metadata.version("mcp")
+async def main() -> None:
+    server: Server[dict[str, object]] = Server("mcp")
     async with stdio_server() as (read_stream, write_stream):
-        async with (
-            ServerSession(
-                read_stream,
-                write_stream,
-                InitializationOptions(
-                    server_name="mcp",
-                    server_version=version,
-                    capabilities=ServerCapabilities(),
-                ),
-            ) as session,
-            write_stream,
-        ):
-            await receive_loop(session)
+        await server.run(read_stream, write_stream, server.create_initialization_options())
 
 
 if __name__ == "__main__":

tests/conftest.py

@@ -1,4 +1,5 @@
 import os
+from collections.abc import Iterator
 
 import pytest
 
@@ -11,7 +12,33 @@
 # suite-wide. Set before logfire is imported anywhere.
 os.environ.setdefault("LOGFIRE_DISTRIBUTED_TRACING", "true")
 
+import opentelemetry.trace  # noqa: E402  (env var must be set before logfire import below)
+from logfire.testing import CaptureLogfire  # noqa: E402
+
+import mcp.shared._otel  # noqa: E402
+
 
 @pytest.fixture
 def anyio_backend():
     return "asyncio"
+
+
+@pytest.fixture(name="capfire")
+def _capfire_isolated(capfire: CaptureLogfire) -> Iterator[CaptureLogfire]:
+    """Override of logfire's `capfire` that scopes the MCP tracer to the test.
+
+    `capfire` installs a real tracer provider, and logfire's proxy machinery
+    mutates the cached `mcp.shared._otel._tracer` to delegate to it for the
+    rest of the process. Without isolation, every subsequent test in the same
+    worker would emit real spans, and `send_raw_request` would inject a real
+    `traceparent` into outbound `_meta`, breaking the interaction-suite
+    snapshots that pin `_meta={}` under a no-op tracer.
+
+    Setup points `_tracer` at the now-live provider so MCP spans record;
+    teardown replaces it with a `NoOpTracer`.
+    """
+    mcp.shared._otel._tracer = opentelemetry.trace.get_tracer_provider().get_tracer("mcp-python-sdk")
+    try:
+        yield capfire
+    finally:
+        mcp.shared._otel._tracer = opentelemetry.trace.NoOpTracer()

tests/issues/test_malformed_input.py

@@ -36,10 +36,13 @@ def finished(self) -> list[ReadableSpan]:
 def spans(capfire: CaptureLogfire) -> Iterator[SpanCapture]:
     """In-memory MCP span capture, cleared before and after each test.
 
-    Backed by the project-level `capfire` override (see `tests/conftest.py`)
-    so there is a single global tracer provider for the suite.
+    Backed by the project-level `capfire` override (see `tests/conftest.py`),
+    which scopes `mcp.shared._otel._tracer` to the test so the real tracer
+    doesn't leak into later tests in the same worker.
     """
     capture = SpanCapture(capfire.exporter)
     capture.clear()
     yield capture
     capture.clear()
+
+

tests/server/conftest.py

@@ -17,9 +17,12 @@
 from mcp.shared.dispatcher import CallOptions
 from mcp.shared.exceptions import NoBackChannelError
 from mcp.types import (
+    LATEST_PROTOCOL_VERSION,
     ClientCapabilities,
     ElicitationCapability,
     EmptyResult,
+    Implementation,
+    InitializeRequestParams,
     ListRootsRequest,
     ListRootsResult,
     PingRequest,
@@ -28,6 +31,14 @@
 )
 
 
+def _client_params(capabilities: ClientCapabilities) -> InitializeRequestParams:
+    return InitializeRequestParams(
+        protocol_version=LATEST_PROTOCOL_VERSION,
+        capabilities=capabilities,
+        client_info=Implementation(name="t", version="0"),
+    )
+
+
 class StubOutbound:
     def __init__(
         self, *, result: dict[str, Any] | None = None, raise_on_send: type[BaseException] | None = None
@@ -190,14 +201,24 @@ def test_connection_check_capability_false_before_initialized():
 )
 def test_check_capability_per_field_branches(have: ClientCapabilities, want: ClientCapabilities, expected: bool):
     conn = Connection(StubOutbound(), has_standalone_channel=True)
-    conn.client_capabilities = have
+    conn.client_params = _client_params(have)
     assert conn.check_capability(want) is expected
 
 
+def test_connection_client_info_and_capabilities_derive_from_client_params():
+    conn = Connection(StubOutbound(), has_standalone_channel=True)
+    assert conn.client_info is None
+    assert conn.client_capabilities is None
+    caps = ClientCapabilities(sampling=SamplingCapability())
+    conn.client_params = _client_params(caps)
+    assert conn.client_info is not None and conn.client_info.name == "t"
+    assert conn.client_capabilities == caps
+
+
 def test_connection_check_capability_true_when_client_declares_it():
     conn = Connection(StubOutbound(), has_standalone_channel=True)
-    conn.client_capabilities = ClientCapabilities(
-        sampling=SamplingCapability(), roots=RootsCapability(list_changed=True)
+    conn.client_params = _client_params(
+        ClientCapabilities(sampling=SamplingCapability(), roots=RootsCapability(list_changed=True))
     )
     conn.initialized.set()
     assert conn.check_capability(ClientCapabilities(sampling=SamplingCapability())) is True