diff --git a/docs/get-started/testing.md b/docs/get-started/testing.md
index 98c671738f..b9211f04d1 100644
--- a/docs/get-started/testing.md
+++ b/docs/get-started/testing.md
@@ -12,18 +12,18 @@ Let's assume you have a simple server with a single tool:
 --8<-- "docs_src/testing/tutorial001.py"
 ```
 
-To run the test below you'll need two extra (development) dependencies:
+Install the development dependencies to run the test on both async backends:
 
 === "uv"
 
     ```bash
-    uv add --dev pytest inline-snapshot
+    uv add --dev pytest inline-snapshot trio
     ```
 
 === "pip"
 
     ```bash
-    pip install pytest inline-snapshot
+    pip install pytest inline-snapshot trio
     ```
 
 !!! info
@@ -45,9 +45,9 @@ from mcp.types import CallToolResult, TextContent
 from server import mcp
 
 
-@pytest.fixture
-def anyio_backend():  # (1)!
-    return "asyncio"
+@pytest.fixture(params=["asyncio", "trio"])
+def anyio_backend(request: pytest.FixtureRequest) -> str:  # (1)!
+    return request.param
 
 
 @pytest.fixture
@@ -69,7 +69,7 @@ async def test_call_add_tool(client: Client):
     )
 ```
 
-1. If you are using `trio`, return `"trio"` instead. See the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on) for the details.
+1. Each test runs once with `asyncio` and once with `trio`. Testing both catches backend-specific assumptions and scheduling races. If your application requires one backend, keep only that name in `params`. See the [anyio documentation](https://anyio.readthedocs.io/en/stable/testing.html#specifying-the-backends-to-run-on) for details.
 2. The fixture yields a connected client. Every test that takes `client` gets a fresh in-memory connection to the same server.
 
 There you go! You can now extend your tests to cover more scenarios.
diff --git a/pyproject.toml b/pyproject.toml
index b2f26da55f..0aa254b25d 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -284,6 +284,8 @@ filterwarnings = [
     # CI and scripts/test set PYTHONWARNDEFAULTENCODING=1, so "error" rejects any text I/O
     # of ours that omits encoding=; pytest-examples' own unguarded text I/O isn't ours.
     "ignore:'encoding' argument not specified:EncodingWarning:pytest_examples",
+    # Trio wraps Linux pidfds in text mode without encoding=; only fileno()/close() are used.
+    "ignore:'encoding' argument not specified:EncodingWarning:trio\\._subprocess$",
 ]
 
 [tool.markdown.lint]
diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py
index 0101e45a76..09e6996dd5 100644
--- a/tests/client/test_stdio.py
+++ b/tests/client/test_stdio.py
@@ -905,9 +905,16 @@ async def test_invalid_utf8_flushed_by_a_dying_server_does_not_break_shutdown(
     abort the drain or surface a UnicodeDecodeError out of the context manager.
     """
     ping = JSONRPCRequest(jsonrpc="2.0", id=1, method="ping")
-    process = FakeProcess(on_stdin_close=lambda: process.exit(0))
+    process = FakeProcess()
     terminated = install_fake_process(monkeypatch, process)
 
+    def exit_when_flushed() -> None:
+        if process.stdin_closed.is_set() and process.pending_stdout_chunks() == 0:
+            process.exit(0)
+
+    process.on_stdin_close = exit_when_flushed
+    process.on_stdout_receive = exit_when_flushed
+
     with anyio.fail_after(5):
         async with stdio_client(FAKE_PARAMS):
             # Park the reader delivering a message nobody receives, then queue
diff --git a/tests/conftest.py b/tests/conftest.py
index 4350b76f78..8e19ec95f6 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -23,9 +23,9 @@
 del _httpcore2
 
 
-@pytest.fixture(scope="session")
-def anyio_backend() -> str:
-    return "asyncio"
+@pytest.fixture(scope="session", params=["asyncio", "trio"])
+def anyio_backend(request: pytest.FixtureRequest) -> str:
+    return request.param
 
 
 @pytest.fixture(autouse=True)
@@ -51,7 +51,7 @@ def blockbuster() -> Iterator[None]:
 
 @pytest.fixture(scope="module", autouse=True)
 async def _module_runner_lease(anyio_backend: str) -> AsyncIterator[None]:
-    """Share one event loop across each module's tests instead of one per test.
+    """Share one event loop per module and backend instead of one per test.
 
     anyio's pytest plugin tears its runner down whenever the last lease is
     released, so with only function-scoped async fixtures every async test
@@ -60,15 +60,15 @@ async def _module_runner_lease(anyio_backend: str) -> AsyncIterator[None]:
     run can transiently exhaust kernel socket buffers — surfacing in CI as
     `OSError: [WinError 10055]` raised from `asyncio.new_event_loop()` before
     an arbitrary test's body even starts. Holding a module-scoped lease caps
-    the churn at one loop per module per xdist worker.
+    the churn at one loop per module and backend per xdist worker.
 
     Modules that parametrize `anyio_backend` or call `trio.run(...)` directly
     must shadow this fixture with a sync no-op: a module-scoped lease cannot
     depend on the function-scoped parameter (pytest raises ScopeMismatch at
     setup), and the lease's live asyncio loop lingers over direct trio runs,
     whose signal handling collides with the loop's wakeup fd on Windows. The
-    lease also makes sniffio report asyncio to the module's sync tests, so a
-    sync test must not call `anyio.run()` itself.
+    lease also makes sniffio report the leased backend to the module's sync
+    tests, so a sync test must not call `anyio.run()` itself.
     """
     yield
 
diff --git a/tests/docs_src/test_asgi.py b/tests/docs_src/test_asgi.py
index d241dc71f9..8653206642 100644
--- a/tests/docs_src/test_asgi.py
+++ b/tests/docs_src/test_asgi.py
@@ -1,6 +1,7 @@
 """`docs/run/asgi.md`: every claim the page makes, proved against the real SDK."""
 
 import inspect
+from importlib import reload
 
 import httpx2
 import pytest
@@ -113,6 +114,7 @@ async def about(request: Request) -> Response:
 
 async def test_the_host_lifespan_enters_the_session_manager() -> None:
     """tutorial002: the host app's lifespan owns `session_manager.run()` and starts and stops cleanly."""
+    reload(tutorial002)
     async with tutorial002.lifespan(tutorial002.app):
         async with Client(tutorial002.mcp) as client:
             result = await client.call_tool("add_note", {"text": "milk"})
@@ -130,6 +132,7 @@ async def test_two_servers_get_two_mounts() -> None:
 
 async def test_one_lifespan_starts_both_session_managers() -> None:
     """tutorial003: a single `AsyncExitStack` lifespan runs both managers; both servers answer."""
+    reload(tutorial003)
     async with tutorial003.lifespan(tutorial003.app):
         async with Client(tutorial003.notes) as client:
             notes_result = await client.call_tool("add_note", {"text": "milk"})
@@ -215,6 +218,7 @@ async def test_the_default_app_is_localhost_only() -> None:
 async def test_the_documented_browser_origin_works_end_to_end() -> None:
     """tutorial005: the page's scenario for real. The public hostname, the browser origin, a
     realistic preflight naming the `Mcp-*` headers, then the actual request."""
+    reload(tutorial005)
     transport = httpx2.ASGITransport(app=tutorial005.app)
     async with tutorial005.lifespan(tutorial005.app):
         async with httpx2.AsyncClient(transport=transport, base_url="https://mcp.example.com") as http:
diff --git a/tests/docs_src/test_context.py b/tests/docs_src/test_context.py
index 617d113b2b..2c41ab1505 100644
--- a/tests/docs_src/test_context.py
+++ b/tests/docs_src/test_context.py
@@ -1,6 +1,7 @@
 """`docs/handlers/context.md`: every claim the page makes, proved against the real SDK."""
 
 import re
+from importlib import reload
 
 import pytest
 from inline_snapshot import snapshot
@@ -62,6 +63,7 @@ async def test_a_context_only_tool_takes_no_arguments() -> None:
 
 async def test_register_a_tool_at_runtime_and_notify_the_client() -> None:
     """tutorial003: `mcp.add_tool` takes effect immediately and `send_tool_list_changed` reaches the client."""
+    reload(tutorial003)
     messages: list[object] = []
 
     async def collect(message: object) -> None:
diff --git a/tests/docs_src/test_deploy.py b/tests/docs_src/test_deploy.py
index 268c7ab74d..442c35be4a 100644
--- a/tests/docs_src/test_deploy.py
+++ b/tests/docs_src/test_deploy.py
@@ -1,5 +1,7 @@
 """`docs/run/deploy.md`: every claim the page makes, proved against the real SDK."""
 
+from importlib import reload
+
 import anyio
 import httpx2
 import pytest
@@ -51,6 +53,7 @@ async def test_the_default_app_rejects_a_real_hostname_before_mcp_runs() -> None
 
 async def test_the_allowlisted_app_serves_its_hostname_and_still_rejects_others() -> None:
     """tutorial001: `allowed_hosts=` opens exactly the hostname you named, and nothing else."""
+    reload(tutorial001)
     transport = httpx2.ASGITransport(app=tutorial001.app)
     async with tutorial001.mcp.session_manager.run():
         async with httpx2.AsyncClient(transport=transport, base_url="https://mcp.example.com") as http:
diff --git a/tests/docs_src/test_identity_assertion.py b/tests/docs_src/test_identity_assertion.py
index 3a15ef94ec..6e172edf67 100644
--- a/tests/docs_src/test_identity_assertion.py
+++ b/tests/docs_src/test_identity_assertion.py
@@ -1,6 +1,7 @@
 """`docs/client/identity-assertion.md`: every claim the page makes, proved against the real SDK."""
 
 import inspect
+from importlib import reload
 from urllib.parse import parse_qsl
 
 import httpx2
@@ -144,6 +145,7 @@ async def test_the_metadata_advertises_the_grant_type_and_the_id_jag_profile() -
 
 async def test_the_whole_grant_is_one_token_request() -> None:
     """The `!!! check`: a 401, the well-known fetch, one `POST /token`, the retry; the subject reaches the tool."""
+    reload(tutorial001)
     mcp = MCPServer(
         "Notes",
         token_verifier=ProviderTokenVerifier(tutorial002.provider),
diff --git a/tests/docs_src/test_legacy_clients.py b/tests/docs_src/test_legacy_clients.py
index 90daf1bd96..e5065ddea1 100644
--- a/tests/docs_src/test_legacy_clients.py
+++ b/tests/docs_src/test_legacy_clients.py
@@ -1,6 +1,7 @@
 """`docs/run/legacy-clients.md`: every claim the page makes, proved against the real SDK."""
 
 import inspect
+from importlib import reload
 
 import httpx2
 import pytest
@@ -115,6 +116,7 @@ async def test_stateless_http_never_mints_a_session() -> None:
 async def test_stateless_http_kills_the_legacy_back_channel_and_only_the_legacy_one() -> None:
     """tutorial002: over the same `stateless_http=True` app, the modern client still gets its answer and
     the legacy client's call fails as the top-level `MCPError` the `!!! check` quotes."""
+    reload(tutorial002)
     async with (
         tutorial002.app.router.lifespan_context(tutorial002.app),
         httpx2.ASGITransport(tutorial002.app) as transport,
diff --git a/tests/docs_src/test_subscriptions.py b/tests/docs_src/test_subscriptions.py
index e2d9bf2c77..dd98eb12a8 100644
--- a/tests/docs_src/test_subscriptions.py
+++ b/tests/docs_src/test_subscriptions.py
@@ -1,6 +1,7 @@
 """`docs/{handlers,client}/subscriptions.md`: every claim the two pages make, proved against the real SDK."""
 
-from collections.abc import Awaitable, Callable
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
 from typing import Any
 
 import anyio
@@ -19,18 +20,14 @@
     tutorial006,
 )
 from mcp import Client
+from mcp.client.subscriptions import Subscription
 from mcp.server.auth.middleware.auth_context import auth_context_var
 from mcp.server.auth.middleware.bearer_auth import AuthenticatedUser
 from mcp.server.auth.provider import AccessToken
-from mcp.server.context import ServerRequestContext
 from mcp.server.lowlevel import Server
 from mcp.server.subscriptions import SUBSCRIPTION_ID_META_KEY, ListenHandler, ToolsListChanged
 from mcp.shared.exceptions import MCPError
 
-_ReadResource = Callable[
-    [ServerRequestContext[Any], types.ReadResourceRequestParams], Awaitable[types.ReadResourceResult]
-]
-
 # See test_index.py for why this is a per-module mark and not a conftest hook.
 pytestmark = [pytest.mark.anyio, pytest.mark.filterwarnings("error::mcp.MCPDeprecationWarning")]
 
@@ -68,24 +65,18 @@ async def wait_for(self, count: int) -> None:
                 await self._arrival.wait()
 
 
-class _Reads:
-    """Counts server-side resource reads so a test can await the Nth refetch."""
+class _Output:
+    """Counts completed prints so tests wait for client output, not server-side reads."""
 
     def __init__(self) -> None:
         self.count = 0
         self._bump = anyio.Event()
 
-    def counting(self, handler: _ReadResource) -> _ReadResource:
-        async def counted(
-            ctx: ServerRequestContext[Any], params: types.ReadResourceRequestParams
-        ) -> types.ReadResourceResult:
-            result = await handler(ctx, params)
-            self.count += 1
-            self._bump.set()
-            self._bump = anyio.Event()
-            return result
-
-        return counted
+    def __call__(self, *values: str | list[str]) -> None:
+        print(*values)
+        self.count += 1
+        self._bump.set()
+        self._bump = anyio.Event()
 
     async def wait_for(self, count: int) -> None:
         with anyio.fail_after(5):
@@ -209,18 +200,34 @@ async def listen() -> None:
 
 
 async def test_follow_board_prints_the_refetched_board_and_the_new_tool_list(
-    capsys: pytest.CaptureFixture[str],
+    capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
 ) -> None:
     """tutorial003: each event drives a refetch - the board reprints, and a tools change reprints the tool names."""
+    output = _Output()
+    monkeypatch.setattr(tutorial003, "print", output, raising=False)
+    listening = anyio.Event()
     async with Client(tutorial001.mcp) as client:
+        listen = client.listen
+
+        @asynccontextmanager
+        async def listen_when_ready(
+            *, tools_list_changed: bool, resource_subscriptions: list[str]
+        ) -> AsyncIterator[Subscription]:
+            async with listen(
+                tools_list_changed=tools_list_changed, resource_subscriptions=resource_subscriptions
+            ) as sub:
+                listening.set()
+                yield sub
+
+        monkeypatch.setattr(client, "listen", listen_when_ready)
         async with anyio.create_task_group() as tg:
             tg.start_soon(tutorial003.follow_board, client)
-            # Let the watcher park on its stream (ack complete) before publishing.
-            await anyio.wait_all_tasks_blocked()
+            with anyio.fail_after(5):
+                await listening.wait()
             await client.call_tool("complete_task", {"board": "sprint", "task": "design"})
-            await anyio.wait_all_tasks_blocked()
+            await output.wait_for(1)
             await client.call_tool("enable_reports", {})
-            await anyio.wait_all_tasks_blocked()
+            await output.wait_for(2)
             tg.cancel_scope.cancel()
 
     printed = capsys.readouterr().out
@@ -243,13 +250,15 @@ def _assert_snapshot_then_current_board(printed: str) -> None:
     assert printed.strip().endswith(FINISHED_BOARD), printed
 
 
+@pytest.mark.parametrize("anyio_backend", [pytest.param("asyncio", id="asyncio")])
 async def test_the_asyncio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
     """tutorial004 (asyncio tab): run_sprint opens the subscription, snapshots the board, then a watcher
     task reprints it while the main flow keeps calling tools.
 
     The example connects over HTTP; the in-memory client here is the maintainer-side stand-in."""
     async with Client(tutorial001.mcp) as client:
-        await tutorial004_asyncio.run_sprint(client)
+        with anyio.fail_after(5):
+            await tutorial004_asyncio.run_sprint(client)
     _assert_snapshot_then_current_board(capsys.readouterr().out)
 
 
@@ -257,14 +266,16 @@ async def test_the_asyncio_watcher_runs_beside_the_main_flow(capsys: pytest.Capt
 async def test_the_trio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
     """tutorial004 (trio tab): the same shape as the asyncio tab, with a nursery owning the watcher."""
     async with Client(tutorial001.mcp) as client:
-        await tutorial004_trio.run_sprint(client)
+        with anyio.fail_after(5):
+            await tutorial004_trio.run_sprint(client)
     _assert_snapshot_then_current_board(capsys.readouterr().out)
 
 
 async def test_the_anyio_watcher_runs_beside_the_main_flow(capsys: pytest.CaptureFixture[str]) -> None:
     """tutorial004 (anyio tab): the same shape again, with a task group owning the watcher."""
     async with Client(tutorial001.mcp) as client:
-        await tutorial004_anyio.run_sprint(client)
+        with anyio.fail_after(5):
+            await tutorial004_anyio.run_sprint(client)
     _assert_snapshot_then_current_board(capsys.readouterr().out)
 
 
@@ -272,16 +283,19 @@ async def test_the_anyio_watcher_runs_beside_the_main_flow(capsys: pytest.Captur
     "anyio_backend",
     [pytest.param(("trio", {"clock": MockClock(autojump_threshold=0)}), id="trio-mockclock")],
 )
-async def test_the_follower_re_listens_after_the_stream_ends(capsys: pytest.CaptureFixture[str]) -> None:
+async def test_the_follower_re_listens_after_the_stream_ends(
+    capsys: pytest.CaptureFixture[str], monkeypatch: pytest.MonkeyPatch
+) -> None:
     """tutorial005: a graceful server close ends one stream; the loop backs off, re-listens, and refetches.
 
     Runs on trio's autojumping MockClock so the loop's backoff sleep takes no wall-clock time.
     """
-    reads = _Reads()
+    output = _Output()
+    monkeypatch.setattr(tutorial005, "print", output, raising=False)
     handler = ListenHandler(tutorial002.bus)
     server = Server(
         "sprint-board",
-        on_read_resource=reads.counting(tutorial002.read_resource),
+        on_read_resource=tutorial002.read_resource,
         on_list_tools=tutorial002.list_tools,
         on_call_tool=tutorial002.call_tool,
         on_subscriptions_listen=handler,
@@ -290,17 +304,16 @@ async def test_the_follower_re_listens_after_the_stream_ends(capsys: pytest.Capt
     async with Client(server) as client:
         async with anyio.create_task_group() as tg:
             tg.start_soon(tutorial005.keep_following, client)
-            # First stream: the entry refetch reads the board, then an event reads it again.
-            await reads.wait_for(1)
+            # First stream: print the entry snapshot, then the board after the event.
+            await output.wait_for(1)
             await client.call_tool("complete_task", {"task": "design"})
-            await reads.wait_for(2)
+            await output.wait_for(2)
 
-            # End that stream gracefully. The loop backs off (the mock clock jumps the
-            # sleep), re-listens, and refetches on entry: that is the third read.
+            # The mock clock jumps the backoff; the third print is the new stream's snapshot.
             handler.close()
-            await reads.wait_for(3)
+            await output.wait_for(3)
             await client.call_tool("complete_task", {"task": "build"})
-            await reads.wait_for(4)
+            await output.wait_for(4)
             tg.cancel_scope.cancel()
 
     printed = capsys.readouterr().out
diff --git a/tests/docs_src/test_troubleshooting.py b/tests/docs_src/test_troubleshooting.py
index 1e1b5e15b8..ab69bb86b3 100644
--- a/tests/docs_src/test_troubleshooting.py
+++ b/tests/docs_src/test_troubleshooting.py
@@ -1,6 +1,7 @@
 """`docs/troubleshooting.md`: every error string the page names, reproduced against the real SDK."""
 
 import logging
+from importlib import reload
 from typing import Any
 
 import httpx2
@@ -147,6 +148,7 @@ async def test_the_default_streamable_http_app_answers_a_real_hostname_with_421(
     caplog: pytest.LogCaptureFixture,
 ) -> None:
     """tutorial003: one 421, three spellings. The page presents all three as the same event."""
+    reload(tutorial003)
     transport = httpx2.ASGITransport(app=tutorial003.app)
     async with tutorial003.mcp.session_manager.run():
         # What curl (or the reverse proxy's access log) shows: the status and the plain-text body.
@@ -169,6 +171,7 @@ async def test_the_default_streamable_http_app_answers_a_real_hostname_with_421(
 
 async def test_an_allowlisted_hostname_connects_and_calls_a_tool() -> None:
     """tutorial004: `transport_security=` names the deployed hostname, and the same client connects."""
+    reload(tutorial004)
     transport = httpx2.ASGITransport(app=tutorial004.app)
     async with tutorial004.mcp.session_manager.run():
         async with httpx2.AsyncClient(transport=transport) as http_client:
@@ -265,6 +268,7 @@ async def test_a_legacy_ctx_elicit_without_a_callback_says_elicitation_not_suppo
 
 async def test_ctx_elicit_over_stateless_http_has_no_back_channel() -> None:
     """tutorial008: `stateless_http=True` leaves the server no channel to send `elicitation/create`."""
+    reload(tutorial008)
     transport = httpx2.ASGITransport(app=tutorial008.app)
     async with tutorial008.mcp.session_manager.run():
         async with httpx2.AsyncClient(transport=transport) as http_client:
diff --git a/tests/issues/test_1363_race_condition_streamable_http.py b/tests/issues/test_1363_race_condition_streamable_http.py
index f98194b7b5..f290dc35b4 100644
--- a/tests/issues/test_1363_race_condition_streamable_http.py
+++ b/tests/issues/test_1363_race_condition_streamable_http.py
@@ -16,12 +16,10 @@
 """
 
 import logging
-import threading
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
 
 import anyio
-import anyio.to_thread
 import httpx2
 import pytest
 from starlette.applications import Starlette
@@ -62,42 +60,6 @@ async def lifespan(app: Starlette) -> AsyncGenerator[None, None]:
     return Starlette(routes=routes, lifespan=lifespan)
 
 
-class ServerThread(threading.Thread):
-    """Thread that runs the ASGI application lifespan in a separate event loop."""
-
-    def __init__(self, app: Starlette):
-        super().__init__(daemon=True)
-        self.app = app
-        self._stop_event = threading.Event()
-        self._ready_event = threading.Event()
-
-    def run(self) -> None:
-        """Run the lifespan in a new event loop."""
-
-        # Create a new event loop for this thread
-        async def run_lifespan():
-            # Use the lifespan context (always present in our tests)
-            lifespan_context = getattr(self.app.router, "lifespan_context", None)
-            assert lifespan_context is not None  # Tests always create apps with lifespan
-            async with lifespan_context(self.app):
-                # Only signal readiness once lifespan startup has completed, i.e. the
-                # session manager's task group exists and requests can be handled.
-                self._ready_event.set()
-                # Wait until stop is requested
-                while not self._stop_event.is_set():
-                    await anyio.sleep(0.1)
-
-        anyio.run(run_lifespan)
-
-    def wait_ready(self, timeout: float = 5.0) -> None:
-        """Block until the lifespan has started; call from a worker thread, not the event loop."""
-        assert self._ready_event.wait(timeout), "server thread did not start its lifespan in time"
-
-    def stop(self) -> None:
-        """Signal the thread to stop."""
-        self._stop_event.set()
-
-
 def check_logs_for_race_condition_errors(caplog: pytest.LogCaptureFixture, test_name: str) -> None:
     """Check logs for ClosedResourceError and other race condition errors.
 
@@ -128,7 +90,7 @@ def check_logs_for_race_condition_errors(caplog: pytest.LogCaptureFixture, test_
 
 
 @pytest.mark.anyio
-async def test_race_condition_invalid_accept_headers(caplog: pytest.LogCaptureFixture):
+async def test_race_condition_invalid_accept_headers(caplog: pytest.LogCaptureFixture) -> None:
     """Test the race condition with invalid Accept headers.
 
     This test reproduces the exact scenario described in issue #1363:
@@ -137,147 +99,117 @@ async def test_race_condition_invalid_accept_headers(caplog: pytest.LogCaptureFi
     - This should trigger the race condition where message_router encounters ClosedResourceError
     """
     app = create_app()
-    server_thread = ServerThread(app)
-    server_thread.start()
-
     try:
-        # Wait for the server thread to enter the lifespan before sending requests
-        await anyio.to_thread.run_sync(server_thread.wait_ready)
-
-        # Suppress WARNING logs (expected validation errors) and capture ERROR logs
-        with caplog.at_level(logging.ERROR):
-            # Test with missing text/event-stream in Accept header
-            async with httpx2.AsyncClient(
-                transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
-            ) as client:
-                response = await client.post(
-                    "/",
-                    json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
-                    headers={
-                        "Accept": "application/json",  # Missing text/event-stream
-                        "Content-Type": "application/json",
-                    },
-                )
-                # Should get 406 Not Acceptable due to missing text/event-stream
-                assert response.status_code == 406
-
-            # Test with missing application/json in Accept header
-            async with httpx2.AsyncClient(
-                transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
-            ) as client:
-                response = await client.post(
-                    "/",
-                    json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
-                    headers={
-                        "Accept": "text/event-stream",  # Missing application/json
-                        "Content-Type": "application/json",
-                    },
-                )
-                # Should get 406 Not Acceptable due to missing application/json
-                assert response.status_code == 406
-
-            # Test with completely invalid Accept header
-            async with httpx2.AsyncClient(
-                transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
-            ) as client:
-                response = await client.post(
-                    "/",
-                    json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
-                    headers={
-                        "Accept": "text/plain",  # Invalid Accept header
-                        "Content-Type": "application/json",
-                    },
-                )
-                # Should get 406 Not Acceptable
-                assert response.status_code == 406
-
-            # Give background tasks time to complete
-            await anyio.sleep(0.2)
-
+        with caplog.at_level(logging.ERROR), anyio.fail_after(5):
+            async with app.router.lifespan_context(app):
+                # Test with missing text/event-stream in Accept header
+                async with httpx2.AsyncClient(
+                    transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
+                ) as client:
+                    response = await client.post(
+                        "/",
+                        json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
+                        headers={
+                            "Accept": "application/json",  # Missing text/event-stream
+                            "Content-Type": "application/json",
+                        },
+                    )
+                    # Should get 406 Not Acceptable due to missing text/event-stream
+                    assert response.status_code == 406
+
+                # Test with missing application/json in Accept header
+                async with httpx2.AsyncClient(
+                    transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
+                ) as client:
+                    response = await client.post(
+                        "/",
+                        json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
+                        headers={
+                            "Accept": "text/event-stream",  # Missing application/json
+                            "Content-Type": "application/json",
+                        },
+                    )
+                    # Should get 406 Not Acceptable due to missing application/json
+                    assert response.status_code == 406
+
+                # Test with completely invalid Accept header
+                async with httpx2.AsyncClient(
+                    transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
+                ) as client:
+                    response = await client.post(
+                        "/",
+                        json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
+                        headers={
+                            "Accept": "text/plain",  # Invalid Accept header
+                            "Content-Type": "application/json",
+                        },
+                    )
+                    # Should get 406 Not Acceptable
+                    assert response.status_code == 406
+
+                # Let the message routers finish before lifespan shutdown cancels them.
+                await anyio.wait_all_tasks_blocked()
     finally:
-        server_thread.stop()
-        server_thread.join(timeout=5.0)
-        # Check logs for race condition errors
         check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_accept_headers")
 
 
 @pytest.mark.anyio
-async def test_race_condition_invalid_content_type(caplog: pytest.LogCaptureFixture):
+async def test_race_condition_invalid_content_type(caplog: pytest.LogCaptureFixture) -> None:
     """Test the race condition with invalid Content-Type headers.
 
     This test reproduces the race condition scenario with Content-Type validation failure.
     """
     app = create_app()
-    server_thread = ServerThread(app)
-    server_thread.start()
-
     try:
-        # Wait for the server thread to enter the lifespan before sending requests
-        await anyio.to_thread.run_sync(server_thread.wait_ready)
-
-        # Suppress WARNING logs (expected validation errors) and capture ERROR logs
-        with caplog.at_level(logging.ERROR):
-            # Test with invalid Content-Type
-            async with httpx2.AsyncClient(
-                transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
-            ) as client:
-                response = await client.post(
-                    "/",
-                    json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
-                    headers={
-                        "Accept": "application/json, text/event-stream",
-                        "Content-Type": "text/plain",  # Invalid Content-Type
-                    },
-                )
-                assert response.status_code == 400
-
-            # Give background tasks time to complete
-            await anyio.sleep(0.2)
-
+        with caplog.at_level(logging.ERROR), anyio.fail_after(5):
+            async with app.router.lifespan_context(app):
+                # Test with invalid Content-Type
+                async with httpx2.AsyncClient(
+                    transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
+                ) as client:
+                    response = await client.post(
+                        "/",
+                        json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
+                        headers={
+                            "Accept": "application/json, text/event-stream",
+                            "Content-Type": "text/plain",  # Invalid Content-Type
+                        },
+                    )
+                    assert response.status_code == 400
+
+                # Let the message router finish before lifespan shutdown cancels it.
+                await anyio.wait_all_tasks_blocked()
     finally:
-        server_thread.stop()
-        server_thread.join(timeout=5.0)
-        # Check logs for race condition errors
         check_logs_for_race_condition_errors(caplog, "test_race_condition_invalid_content_type")
 
 
 @pytest.mark.anyio
-async def test_race_condition_message_router_async_for(caplog: pytest.LogCaptureFixture):
+async def test_race_condition_message_router_async_for(caplog: pytest.LogCaptureFixture) -> None:
     """Uses json_response=True to trigger the `if self.is_json_response_enabled` branch,
     which reproduces the ClosedResourceError when message_router is suspended
     in async for loop while transport cleanup closes streams concurrently.
     """
     app = create_app(json_response=True)
-    server_thread = ServerThread(app)
-    server_thread.start()
-
     try:
-        # Wait for the server thread to enter the lifespan before sending requests
-        await anyio.to_thread.run_sync(server_thread.wait_ready)
-
-        # Suppress WARNING logs (expected validation errors) and capture ERROR logs
-        with caplog.at_level(logging.ERROR):
-            # Use httpx2.ASGITransport to test the ASGI app directly
-            async with httpx2.AsyncClient(
-                transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
-            ) as client:
-                # Send a valid initialize request
-                response = await client.post(
-                    "/",
-                    json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
-                    headers={
-                        "Accept": "application/json, text/event-stream",
-                        "Content-Type": "application/json",
-                    },
-                )
-                # Should get a successful response
-                assert response.status_code in (200, 201)
-
-            # Give background tasks time to complete
-            await anyio.sleep(0.2)
-
+        with caplog.at_level(logging.ERROR), anyio.fail_after(5):
+            async with app.router.lifespan_context(app):
+                # Use httpx2.ASGITransport to test the ASGI app directly
+                async with httpx2.AsyncClient(
+                    transport=httpx2.ASGITransport(app=app), base_url="http://testserver", timeout=5.0
+                ) as client:
+                    # Send a valid initialize request
+                    response = await client.post(
+                        "/",
+                        json={"jsonrpc": "2.0", "method": "initialize", "id": 1, "params": {}},
+                        headers={
+                            "Accept": "application/json, text/event-stream",
+                            "Content-Type": "application/json",
+                        },
+                    )
+                    # Should get a successful response
+                    assert response.status_code in (200, 201)
+
+                # Let the message router finish before lifespan shutdown cancels it.
+                await anyio.wait_all_tasks_blocked()
     finally:
-        server_thread.stop()
-        server_thread.join(timeout=5.0)
-        # Check logs for race condition errors in message router
         check_logs_for_race_condition_errors(caplog, "test_race_condition_message_router_async_for")
diff --git a/tests/server/test_sse_security.py b/tests/server/test_sse_security.py
index 7e84428600..417e7d8d9f 100644
--- a/tests/server/test_sse_security.py
+++ b/tests/server/test_sse_security.py
@@ -9,8 +9,6 @@
 import sse_starlette.sse
 from mcp_types import JSONRPCRequest, JSONRPCResponse
 from starlette.applications import Starlette
-from starlette.requests import Request
-from starlette.responses import Response
 from starlette.routing import Mount, Route
 from starlette.types import Message, Receive, Scope, Send
 
@@ -45,20 +43,17 @@ def sse_security_client(security_settings: TransportSecuritySettings | None = No
     server = Server(SERVER_NAME)
     sse_transport = SseServerTransport("/messages/", security_settings)
 
-    async def handle_sse(request: Request) -> Response:
-        try:
-            async with sse_transport.connect_sse(request.scope, request.receive, request._send) as (read, write):
-                await server.run(read, write, server.create_initialization_options())
-        except ValueError as e:
-            # Validation error was already handled inside connect_sse, which sent the rejection
-            # response itself; its non-empty body checkpoints, so the test reads the rejection
-            # status before the trailing Response() below sends a second response start.
-            logger.debug(f"SSE connection failed validation: {e}")
-        return Response()
+    class SSEApp:
+        async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+            try:
+                async with sse_transport.connect_sse(scope, receive, send) as (read, write):
+                    await server.run(read, write, server.create_initialization_options())
+            except ValueError:
+                logger.exception("SSE connection failed validation")
 
     app = Starlette(
         routes=[
-            Route("/sse", endpoint=handle_sse),
+            Route("/sse", endpoint=SSEApp()),
             Mount("/messages/", app=sse_transport.handle_post_message),
         ]
     )