Replace subprocess tests with in-process async servers (#2006)

* Use anyio as testing backend

* Remove asyncio markers

* Update streamable http tests

* Replace all subprocess tests

* Replace anyio task groups with asyncio context managers in tests

- Convert run_server_async from anyio task group pattern to asyncio.create_task with async context manager
- Remove task_group fixture from conftest
- Update all test fixtures to use async with run_server_async pattern
- Remove TaskGroup imports from all test files
- Tests now work with pytest-asyncio instead of pytest-anyio

* Update test_github_provider_integration.py

Author: jlowin

docs/development/tests.mdx

@@ -297,7 +297,53 @@ async def test_database_tool():
 
 ### Testing Network Transports
 
-While in-memory testing covers most unit testing needs, you'll occasionally need to test actual network transports. Use the `run_server_in_process` utility to spawn a server in a separate process for testing:
+While in-memory testing covers most unit testing needs, you'll occasionally need to test actual network transports like HTTP or SSE. FastMCP provides two approaches: in-process async servers using AnyIO task groups (preferred), and separate subprocess servers (for special cases).
+
+#### In-Process Network Testing (Preferred)
+
+For most network transport tests, use `run_server_async` with AnyIO task groups. This runs the server as a task in the same process, providing fast, deterministic tests with full debugger support:
+
+```python
+import pytest
+from anyio.abc import TaskGroup
+from fastmcp import FastMCP, Client
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.utilities.tests import run_server_async
+
+def create_test_server() -> FastMCP:
+    """Create a test server instance."""
+    server = FastMCP("TestServer")
+    
+    @server.tool
+    def greet(name: str) -> str:
+        return f"Hello, {name}!"
+    
+    return server
+
+@pytest.fixture
+async def http_server(task_group: TaskGroup) -> str:
+    """Start server in-process using task group."""
+    server = create_test_server()
+    url = await run_server_async(task_group, server, transport="http")
+    return url
+
+async def test_http_transport(http_server: str):
+    """Test actual HTTP transport behavior."""
+    async with Client(
+        transport=StreamableHttpTransport(http_server)
+    ) as client:
+        result = await client.ping()
+        assert result is True
+        
+        greeting = await client.call_tool("greet", {"name": "World"})
+        assert greeting.data == "Hello, World!"
+```
+
+The `task_group` fixture is provided globally by `conftest.py` and automatically handles server lifecycle and cleanup. This approach is faster than subprocess-based testing and provides better error messages.
+
+#### Subprocess Testing (Special Cases)
+
+For tests that require complete process isolation (like STDIO transport or testing subprocess behavior), use `run_server_in_process`:
 
 ```python
 import pytest
@@ -328,12 +374,9 @@ async def test_http_transport(http_server: str):
     ) as client:
         result = await client.ping()
         assert result is True
-        
-        greeting = await client.call_tool("greet", {"name": "World"})
-        assert greeting.data == "Hello, World!"
 ```
 
-The `run_server_in_process` utility handles server lifecycle, port allocation, and cleanup automatically. This pattern is essential for testing transport-specific behavior like timeouts, headers, and authentication. Note that FastMCP often uses the `client_process` marker to isolate tests that spawn processes, as they can create contention in CI.
+The `run_server_in_process` utility handles server lifecycle, port allocation, and cleanup automatically. Use this only when subprocess isolation is truly necessary, as it's slower and harder to debug than in-process testing. FastMCP uses the `client_process` marker to isolate these tests in CI.
 
 ### Documentation Testing
 

pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "openapi-core>=0.19.5",
     "py-key-value-aio[disk,memory]>=0.2.2,<0.3.0",
     "websockets>=15.0.1",
+    "pytest-asyncio>=1.2.0",
 ]
 
 requires-python = ">=3.10"
@@ -59,7 +60,6 @@ dev = [
     "pyinstrument>=5.0.2",
     "pyperclip>=1.9.0",
     "pytest>=8.3.3",
-    "pytest-asyncio>=0.23.5",
     "pytest-cov>=6.1.1",
     "pytest-env>=1.1.5",
     "pytest-flakefinder",
@@ -99,8 +99,6 @@ fallback-version = "0.0.0"
 
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
-asyncio_default_fixture_loop_scope = "session"
-asyncio_default_test_loop_scope = "session"
 # filterwarnings = ["error::DeprecationWarning"]
 timeout = 5
 env = [

src/fastmcp/server/context.py

@@ -6,7 +6,6 @@
 import logging
 import warnings
 import weakref
-from asyncio.locks import Lock
 from collections.abc import Generator, Mapping, Sequence
 from contextlib import contextmanager
 from contextvars import ContextVar, Token
@@ -15,6 +14,7 @@
 from logging import Logger
 from typing import Any, Literal, cast, get_origin, overload
 
+import anyio
 from mcp import LoggingLevel, ServerSession
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import request_ctx
@@ -61,7 +61,7 @@
 
 T = TypeVar("T", default=Any)
 _current_context: ContextVar[Context | None] = ContextVar("context", default=None)  # type: ignore[assignment]
-_flush_lock: Lock = asyncio.Lock()
+_flush_lock = anyio.Lock()
 
 
 @dataclass

src/fastmcp/server/middleware/rate_limiting.py

@@ -1,11 +1,11 @@
 """Rate limiting middleware for protecting FastMCP servers from abuse."""
 
-import asyncio
 import time
 from collections import defaultdict, deque
 from collections.abc import Callable
 from typing import Any
 
+import anyio
 from mcp import McpError
 from mcp.types import ErrorData
 
@@ -33,7 +33,7 @@ def __init__(self, capacity: int, refill_rate: float):
         self.refill_rate = refill_rate
         self.tokens = capacity
         self.last_refill = time.time()
-        self._lock = asyncio.Lock()
+        self._lock = anyio.Lock()
 
     async def consume(self, tokens: int = 1) -> bool:
         """Try to consume tokens from the bucket.
@@ -71,7 +71,7 @@ def __init__(self, max_requests: int, window_seconds: int):
         self.max_requests = max_requests
         self.window_seconds = window_seconds
         self.requests = deque()
-        self._lock = asyncio.Lock()
+        self._lock = anyio.Lock()
 
     async def is_allowed(self) -> bool:
         """Check if a request is allowed."""

src/fastmcp/utilities/tests.py

@@ -5,8 +5,8 @@
 import multiprocessing
 import socket
 import time
-from collections.abc import Callable, Generator
-from contextlib import contextmanager
+from collections.abc import AsyncGenerator, Callable, Generator
+from contextlib import asynccontextmanager, contextmanager
 from typing import TYPE_CHECKING, Any, Literal
 from urllib.parse import parse_qs, urlparse
 
@@ -140,6 +140,88 @@ def run_server_in_process(
             raise RuntimeError("Server process failed to terminate even after kill")
 
 
+@asynccontextmanager
+async def run_server_async(
+    server: FastMCP,
+    port: int | None = None,
+    transport: Literal["http", "streamable-http", "sse"] = "http",
+    path: str = "/mcp",
+    host: str = "127.0.0.1",
+) -> AsyncGenerator[str, None]:
+    """
+    Start a FastMCP server as an asyncio task for in-process async testing.
+
+    This is the recommended way to test FastMCP servers. It runs the server
+    as an async task in the same process, eliminating subprocess coordination,
+    sleeps, and cleanup issues.
+
+    Args:
+        server: FastMCP server instance
+        port: Port to bind to (default: find available port)
+        transport: Transport type ("http", "streamable-http", or "sse")
+        path: URL path for the server (default: "/mcp")
+        host: Host to bind to (default: "127.0.0.1")
+
+    Yields:
+        Server URL string
+
+    Example:
+        ```python
+        import pytest
+        from fastmcp import FastMCP, Client
+        from fastmcp.client.transports import StreamableHttpTransport
+        from fastmcp.utilities.tests import run_server_async
+
+        @pytest.fixture
+        async def server():
+            mcp = FastMCP("test")
+
+            @mcp.tool()
+            def greet(name: str) -> str:
+                return f"Hello, {name}!"
+
+            async with run_server_async(mcp) as url:
+                yield url
+
+        async def test_greet(server: str):
+            async with Client(StreamableHttpTransport(server)) as client:
+                result = await client.call_tool("greet", {"name": "World"})
+                assert result.content[0].text == "Hello, World!"
+        ```
+    """
+    import asyncio
+
+    if port is None:
+        port = find_available_port()
+
+    # Wait a tiny bit for the port to be released if it was just used
+    await asyncio.sleep(0.01)
+
+    # Start server as a background task
+    server_task = asyncio.create_task(
+        server.run_http_async(
+            host=host,
+            port=port,
+            transport=transport,
+            path=path,
+            show_banner=False,
+        )
+    )
+
+    # Give the server a moment to start
+    await asyncio.sleep(0.1)
+
+    try:
+        yield f"http://{host}:{port}{path}"
+    finally:
+        # Cleanup: cancel the task
+        server_task.cancel()
+        try:
+            await server_task
+        except asyncio.CancelledError:
+            pass
+
+
 @contextmanager
 def caplog_for_fastmcp(caplog):
     """Context manager to capture logs from FastMCP loggers even when propagation is disabled."""

tests/cli/test_mcp_server_config_integration.py

@@ -84,7 +84,6 @@ def test_detect_test_fastmcp_json(self, tmp_path):
 class TestConfigWithClient:
     """Test fastmcp.json configuration with client connections."""
 
-    @pytest.mark.asyncio
     async def test_config_server_with_client(self, server_with_config):
         """Test that a server loaded from config works with a client."""
         # Load the config

tests/cli/test_server_args.py

@@ -11,7 +11,6 @@
 class TestServerArguments:
     """Test passing arguments to servers."""
 
-    @pytest.mark.asyncio
     async def test_server_with_argparse(self, tmp_path):
         """Test a server that uses argparse with command line arguments."""
         server_file = tmp_path / "argparse_server.py"
@@ -53,7 +52,6 @@ def get_config() -> dict:
         tools = await server.get_tools()
         assert "get_config" in tools
 
-    @pytest.mark.asyncio
     async def test_server_with_no_args(self, tmp_path):
         """Test a server that uses argparse with no arguments (defaults)."""
         server_file = tmp_path / "default_server.py"
@@ -79,7 +77,6 @@ async def test_server_with_no_args(self, tmp_path):
 
         assert server.name == "DefaultName"
 
-    @pytest.mark.asyncio
     async def test_server_with_sys_argv_access(self, tmp_path):
         """Test a server that directly accesses sys.argv."""
         server_file = tmp_path / "sysargv_server.py"
@@ -112,7 +109,6 @@ async def test_server_with_sys_argv_access(self, tmp_path):
 
         assert server.name == "DirectServer"
 
-    @pytest.mark.asyncio
     async def test_config_server_example(self):
         """Test the actual config_server.py example."""
         # Find the examples directory

tests/client/auth/test_oauth_client.py

@@ -1,4 +1,3 @@
-from collections.abc import Generator
 from urllib.parse import urlparse
 
 import httpx
@@ -9,7 +8,8 @@
 from fastmcp.server.auth.auth import ClientRegistrationOptions
 from fastmcp.server.auth.providers.in_memory import InMemoryOAuthProvider
 from fastmcp.server.server import FastMCP
-from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
+from fastmcp.utilities.http import find_available_port
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_async
 
 
 def fastmcp_server(issuer_url: str):
@@ -35,31 +35,27 @@ def get_test_resource() -> str:
     return server
 
 
-def run_server(host: str, port: int, **kwargs) -> None:
-    fastmcp_server(f"http://{host}:{port}").run(host=host, port=port, **kwargs)
-
-
 @pytest.fixture
-def streamable_http_server() -> Generator[str, None, None]:
-    with run_server_in_process(run_server, transport="http") as url:
-        yield f"{url}/mcp"
+async def streamable_http_server():
+    """Start OAuth-enabled server."""
+    port = find_available_port()
+    server = fastmcp_server(f"http://127.0.0.1:{port}")
+    async with run_server_async(server, port=port, transport="http") as url:
+        yield url
 
 
-@pytest.fixture()
+@pytest.fixture
 def client_unauthorized(streamable_http_server: str) -> Client:
     return Client(transport=StreamableHttpTransport(streamable_http_server))
 
 
-@pytest.fixture()
-def client_with_headless_oauth(
-    streamable_http_server: str,
-) -> Generator[Client, None, None]:
+@pytest.fixture
+def client_with_headless_oauth(streamable_http_server: str) -> Client:
     """Client with headless OAuth that bypasses browser interaction."""
-    client = Client(
+    return Client(
         transport=StreamableHttpTransport(streamable_http_server),
         auth=HeadlessOAuth(mcp_url=streamable_http_server),
     )
-    yield client
 
 
 async def test_unauthorized(client_unauthorized: Client):