diff --git a/.github/actions/run-pytest/action.yml b/.github/actions/run-pytest/action.yml index b7e5509e5..e7ee2b8ec 100644 --- a/.github/actions/run-pytest/action.yml +++ b/.github/actions/run-pytest/action.yml @@ -19,7 +19,7 @@ runs: MAX_PROCS="2" EXTRA_FLAGS="" elif [ "${{ inputs.test-type }}" == "client_process" ]; then - MARKER="client_process" + MARKER="client_process or subprocess_heavy" TIMEOUT="5" MAX_PROCS="0" EXTRA_FLAGS="-x" @@ -29,14 +29,20 @@ runs: MAX_PROCS="0" EXTRA_FLAGS="-x" else - MARKER="not integration and not client_process and not conformance" + MARKER="not integration and not client_process and not subprocess_heavy and not conformance" TIMEOUT="5" MAX_PROCS="4" EXTRA_FLAGS="" fi + # Windows previously ran serially: parallel workers crashed intermittently + # when many tests spawned stdio subprocesses (#2715, reverted in #2726). + # Most of those tests now run in-memory, but tests that spawn a fresh + # interpreter importing all of FastMCP still crash xdist workers on the + # 2-core Windows runners. They carry the subprocess_heavy marker and run + # in the serial client_process step instead. PARALLEL_FLAGS="" - if [ "$MAX_PROCS" != "0" ] && [ "${{ runner.os }}" != "Windows" ]; then + if [ "$MAX_PROCS" != "0" ]; then PARALLEL_FLAGS="--numprocesses auto --maxprocesses $MAX_PROCS --dist worksteal" fi diff --git a/.github/workflows/run-tests.yml b/.github/workflows/run-tests.yml index 1942b80d5..7db2b5865 100644 --- a/.github/workflows/run-tests.yml +++ b/.github/workflows/run-tests.yml @@ -48,7 +48,7 @@ jobs: - name: Run unit tests uses: ./.github/actions/run-pytest - - name: Run client process tests + - name: Run serial subprocess tests uses: ./.github/actions/run-pytest with: test-type: client_process @@ -69,7 +69,7 @@ jobs: - name: Run unit tests uses: ./.github/actions/run-pytest - - name: Run client process tests + - name: Run serial subprocess tests uses: ./.github/actions/run-pytest with: test-type: client_process diff --git a/.github/workflows/run-upgrade-checks.yml b/.github/workflows/run-upgrade-checks.yml index 3be7b8182..485cf1919 100644 --- a/.github/workflows/run-upgrade-checks.yml +++ b/.github/workflows/run-upgrade-checks.yml @@ -67,7 +67,7 @@ jobs: - name: Run unit tests uses: ./.github/actions/run-pytest - - name: Run client process tests + - name: Run serial subprocess tests uses: ./.github/actions/run-pytest with: test-type: client_process diff --git a/docs/development/tests.mdx b/docs/development/tests.mdx index 79131802c..bbdbd972a 100644 --- a/docs/development/tests.mdx +++ b/docs/development/tests.mdx @@ -299,22 +299,19 @@ 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 like HTTP or SSE. FastMCP provides two approaches: in-process async servers (preferred), and separate subprocess servers (for special cases). +In-memory testing covers most unit testing needs, but some behavior only exists over HTTP: middleware, authentication, session management, header handling, and SSE streaming. To test those, serve your server over HTTP with `asgi_client`. -#### In-Process Network Testing (Preferred) +#### Testing Over HTTP - + -For most network transport tests, use `run_server_async` as an async context manager. This runs the server as a task in the same process, providing fast, deterministic tests with full debugger support: +`asgi_client` builds your server's real Starlette app, starts its lifespan, and hands you a connected `Client` that talks to it over the full HTTP stack. The one thing it skips is the socket: requests are dispatched straight into the ASGI application on the current event loop, so there is no port to bind, no uvicorn to start, and no connection to negotiate. Everything else — middleware, authentication, session management, SSE framing — runs exactly as it does in production. ```python -import pytest -from fastmcp import FastMCP, Client -from fastmcp.client.transports import StreamableHttpTransport -from fastmcp.utilities.tests import run_server_async +from fastmcp import FastMCP +from fastmcp.utilities.tests import asgi_client def create_test_server() -> FastMCP: - """Create a test server instance.""" server = FastMCP("TestServer") @server.tool @@ -323,26 +320,84 @@ def create_test_server() -> FastMCP: return server -@pytest.fixture -async def http_server() -> str: - """Start server in-process for testing.""" - server = create_test_server() - async with run_server_async(server) as url: - yield 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 - +async def test_greet_over_http(): + async with asgi_client(create_test_server()) as client: greeting = await client.call_tool("greet", {"name": "World"}) assert greeting.data == "Hello, World!" ``` -The `run_server_async` context manager automatically handles server lifecycle and cleanup. This approach is faster than subprocess-based testing and provides better error messages. +Pass `transport="sse"` to exercise the SSE app instead of streamable HTTP, `path=` to serve on a custom path, `headers=` and `auth=` to configure the client's requests, and any other keyword argument to configure the `Client` itself. + +```python +async def test_tenant_header_is_visible_to_tools(): + async with asgi_client( + create_test_server(), + headers={"X-Tenant-ID": "acme"}, + timeout=5, + ) as client: + await client.ping() +``` + +#### Sharing One Server Across Tests + +When several tests share a server but each needs its own client, use `asgi_server` in a fixture. It yields an `ASGIServer`, whose `client()` method produces a fresh client — with its own session — on demand. + +```python +import pytest +from fastmcp import FastMCP +from fastmcp.utilities.tests import ASGIServer, asgi_server + +@pytest.fixture +async def http_server(): + server = FastMCP("TestServer") + + @server.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + async with asgi_server(server) as running_server: + yield running_server + +async def test_greet(http_server: ASGIServer): + async with http_server.client() as client: + greeting = await client.call_tool("greet", {"name": "World"}) + assert greeting.data == "Hello, World!" + +async def test_sessions_are_isolated(http_server: ASGIServer): + async with http_server.client() as first, http_server.client() as second: + assert await first.ping() is True + assert await second.ping() is True +``` + +For assertions about raw HTTP — status codes, response headers, metadata endpoints — `http_client()` returns an `httpx.AsyncClient` bound to the same app. Because nothing is listening on the network, this is the only way to make raw requests; a plain `httpx.AsyncClient()` cannot reach the server. + +```python +async def test_unauthenticated_request_is_rejected(http_server: ASGIServer): + async with http_server.http_client() as http: + response = await http.post(http_server.url, json={"jsonrpc": "2.0", "id": 1}) + assert response.status_code in (400, 401) +``` + +If you need to build the client transport yourself, `transport()` returns a `StreamableHttpTransport` or `SSETransport` already wired to the in-process app. + +#### Testing on a Real Port + + + +`run_server_async` starts a real uvicorn server on a real TCP port as a task in the current process and yields its URL. Reach for it only when the subject of the test is the network itself — real sockets, TLS, or a server that must be reachable by something other than an in-process client. + +```python +from fastmcp import FastMCP, Client +from fastmcp.utilities.tests import run_server_async + +async def test_server_binds_a_real_port(): + server = FastMCP("TestServer") + + async with run_server_async(server) as url: + assert url.startswith("http://127.0.0.1:") + async with Client(url) as client: + assert await client.ping() is True +``` #### Subprocess Testing (Special Cases) diff --git a/fastmcp_slim/fastmcp/utilities/asgi_transport.py b/fastmcp_slim/fastmcp/utilities/asgi_transport.py new file mode 100644 index 000000000..cb8ee1ab4 --- /dev/null +++ b/fastmcp_slim/fastmcp/utilities/asgi_transport.py @@ -0,0 +1,322 @@ +"""An in-process, full-duplex HTTP transport for driving ASGI applications from httpx. + +Ported from the MCP Python SDK's test suite (`tests/interaction/transports/_bridge.py`, +MIT licensed). + +`httpx2.ASGITransport` runs the application to completion and only then hands the buffered +response to the caller, so a server that streams its response — as the streamable HTTP +transport's SSE responses do — can never converse with the client mid-request: a +server-initiated request nested inside a still-open call deadlocks. +`StreamingASGITransport` removes that limitation by running the application as a background +task and forwarding every `http.response.body` chunk to the client the moment it is sent. +Everything happens on the one event loop: no sockets, no threads, no sleeps. + +The behavioural contract: + +- The request body is buffered before the application is invoked (MCP requests are small + JSON documents); the response streams chunk by chunk. +- Closing the response — or the whole client — delivers `http.disconnect` to the + application, exactly as a real server sees when its peer goes away. +- An exception the application raises before sending `http.response.start` fails the + originating request with that same exception. After the response has started, a failure + is visible to the client only through the response itself (status code, truncated body) — + the same signal a real server over a real socket would give. + +The transport owns an anyio task group for the application tasks; it is opened and closed by +`httpx2.AsyncClient`'s own context manager, so the client must be used as a context manager. +Closing the transport cancels every running application task by default; set +`cancel_on_close=False` to wait for the application's own disconnect handling instead, which +is what the legacy SSE transport relies on for resource cleanup. +""" + +from __future__ import annotations + +import asyncio +import math +from collections.abc import AsyncIterator +from contextlib import asynccontextmanager +from types import TracebackType + +import anyio +import anyio.abc +import httpx2 +from anyio.streams.memory import MemoryObjectReceiveStream +from starlette.types import ASGIApp, Message, Scope + + +class _StreamingResponseBody(httpx2.AsyncByteStream): + """A response body that yields chunks as the application produces them. + + Closing it tells the application the client has gone away (`http.disconnect`), + mirroring a peer that drops the connection mid-response. + """ + + def __init__( + self, + chunks: MemoryObjectReceiveStream[bytes], + client_disconnected: anyio.Event, + ) -> None: + self._chunks = chunks + self._client_disconnected = client_disconnected + + async def __aiter__(self) -> AsyncIterator[bytes]: + async for chunk in self._chunks: + yield chunk + + async def aclose(self) -> None: + self._client_disconnected.set() + await self._chunks.aclose() + + +class StreamingASGITransport(httpx2.AsyncBaseTransport): + """Drive an ASGI application in-process, streaming each response as it is produced. + + This is an `httpx2` transport, so it plugs into anything that accepts an + `httpx2.AsyncClient` — including FastMCP's client transports via their + `httpx_client_factory` argument. + + Args: + app: The ASGI application to drive (e.g. `FastMCP.http_app()`). + cancel_on_close: When True (the default), closing the transport cancels every + application task still running, so harness teardown can never hang. Set to + False to wait for the application's own disconnect handling to complete + instead, which the legacy SSE server transport relies on for cleanup. + + Example: + Drive a FastMCP server's real HTTP app with no sockets: + ```python + import httpx2 + from fastmcp import FastMCP + from fastmcp.utilities.asgi_transport import StreamingASGITransport + + mcp = FastMCP("test") + app = mcp.http_app(transport="http") + + async with app.router.lifespan_context(app): + transport = StreamingASGITransport(app) + async with httpx2.AsyncClient( + transport=transport, base_url="http://testserver" + ) as client: + response = await client.get("/mcp") + ``` + """ + + _task_group: anyio.abc.TaskGroup + + def __init__(self, app: ASGIApp, *, cancel_on_close: bool = True) -> None: + self._app = app + self._cancel_on_close = cancel_on_close + + async def __aenter__(self) -> StreamingASGITransport: + self._task_group = anyio.create_task_group() + await self._task_group.__aenter__() + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None = None, + exc_value: BaseException | None = None, + traceback: TracebackType | None = None, + ) -> None: + # httpx closes every streamed response before closing the transport, so by now each + # application task has been delivered `http.disconnect`. Either cancel immediately, + # or wait for the application's own disconnect handling to unwind. + if self._cancel_on_close: + self._task_group.cancel_scope.cancel() + await self._task_group.__aexit__(exc_type, exc_value, traceback) + + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: + if not isinstance(request.stream, httpx2.AsyncByteStream): + raise TypeError( + "StreamingASGITransport requires an async request stream; " + f"got {type(request.stream).__name__}." + ) + request_body = b"".join([chunk async for chunk in request.stream]) + + scope: Scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": request.method, + "scheme": request.url.scheme, + "path": request.url.path, + "raw_path": request.url.raw_path.split(b"?", maxsplit=1)[0], + "query_string": request.url.query, + "root_path": "", + "headers": [(name.lower(), value) for name, value in request.headers.raw], + "server": (request.url.host, request.url.port), + "client": ("127.0.0.1", 1234), + } + + request_delivered = False + start_received = False + client_disconnected = anyio.Event() + response_started = anyio.Event() + response_status = 0 + response_headers: list[tuple[bytes, bytes]] = [] + application_error: Exception | None = None + chunk_writer, chunk_reader = anyio.create_memory_object_stream[bytes](math.inf) + + async def receive_request() -> Message: + nonlocal request_delivered + if not request_delivered: + request_delivered = True + return { + "type": "http.request", + "body": request_body, + "more_body": False, + } + await client_disconnected.wait() + return {"type": "http.disconnect"} + + async def send_response(message: Message) -> None: + nonlocal response_status, response_headers, start_received + if message["type"] == "http.response.start": + start_received = True + response_status = message["status"] + response_headers = list(message.get("headers", [])) + response_started.set() + return + if message["type"] != "http.response.body": + raise RuntimeError(f"Unexpected ASGI message type: {message['type']}") + body: bytes = message.get("body", b"") + if body: + await chunk_writer.send(body) + if not message.get("more_body", False): + await chunk_writer.aclose() + + async def run_application() -> None: + nonlocal application_error + try: + await self._app(scope, receive_request, send_response) + except Exception as exc: + # The bridge is the application's outermost boundary: a crash must fail the + # originating request (or show up in the already-started response), never + # tear down the task group shared with every other in-flight request. + application_error = exc + finally: + response_started.set() + await chunk_writer.aclose() + + self._task_group.start_soon(run_application) + try: + await response_started.wait() + # Only a failure *before* the start message can fail the request. Once the + # response has started the client sees the failure as a truncated body, which + # is the same signal a real server over a real socket would give. + if application_error is not None and not start_received: + raise application_error + except BaseException: + # No response will be built, so close the reader the response body would have + # owned and tell the application its peer has gone away. + client_disconnected.set() + await chunk_reader.aclose() + raise + return httpx2.Response( + status_code=response_status, + headers=response_headers, + stream=_StreamingResponseBody(chunk_reader, client_disconnected), + request=request, + ) + + +@asynccontextmanager +async def run_asgi_lifespan(app: ASGIApp) -> AsyncIterator[None]: + """Run an ASGI application's lifespan, driving the protocol as a real server does. + + The application's lifespan runs inside a dedicated task for the whole duration of + the context. This matters because a lifespan typically owns cancel scopes and task + groups — anyio requires those to be exited by the task that entered them, which + rules out entering the lifespan on one task and leaving it on another (as a pytest + fixture's setup and teardown phases may do). + + Args: + app: The ASGI application whose lifespan should run. + + Raises: + RuntimeError: If the application reports `lifespan.startup.failed`, or reports + `lifespan.shutdown.failed` (or crashes during shutdown) while the context + body itself completed successfully. A failure inside the body takes + precedence and propagates unchanged. + """ + receive_queue: asyncio.Queue[Message] = asyncio.Queue() + startup_complete: asyncio.Future[None] = asyncio.get_running_loop().create_future() + shutdown_complete: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + async def receive() -> Message: + return await receive_queue.get() + + async def send(message: Message) -> None: + if message["type"] == "lifespan.startup.complete": + if not startup_complete.done(): + startup_complete.set_result(None) + elif message["type"] == "lifespan.startup.failed": + if not startup_complete.done(): + startup_complete.set_exception( + RuntimeError( + f"ASGI application startup failed: {message.get('message', '')}" + ) + ) + elif message["type"] == "lifespan.shutdown.complete": + if not shutdown_complete.done(): + shutdown_complete.set_result(None) + elif message["type"] == "lifespan.shutdown.failed": + if not shutdown_complete.done(): + shutdown_complete.set_exception( + RuntimeError( + "ASGI application shutdown failed: " + f"{message.get('message', '')}" + ) + ) + + async def run_lifespan() -> None: + scope: Scope = {"type": "lifespan", "asgi": {"version": "3.0"}} + try: + await app(scope, receive, send) + except BaseException as exc: + # The app died without completing the handshake; surface that to whichever + # side is still waiting rather than hanging. + if not startup_complete.done(): + startup_complete.set_exception(exc) + if not shutdown_complete.done(): + shutdown_complete.set_exception(exc) + raise + else: + if not startup_complete.done(): + startup_complete.set_exception( + RuntimeError("ASGI application exited before completing startup") + ) + if not shutdown_complete.done(): + shutdown_complete.set_result(None) + + task = asyncio.create_task(run_lifespan()) + await receive_queue.put({"type": "lifespan.startup"}) + try: + await startup_complete + except BaseException: + task.cancel() + with anyio.CancelScope(shield=True): + await asyncio.gather(task, return_exceptions=True) + raise + + body_failed = False + try: + yield + except BaseException: + body_failed = True + raise + finally: + await receive_queue.put({"type": "lifespan.shutdown"}) + with anyio.CancelScope(shield=True): + results = await asyncio.gather( + shutdown_complete, task, return_exceptions=True + ) + # A harness must surface a broken teardown rather than swallow it — but never at + # the cost of masking the failure the body already raised, which is the one the + # caller actually needs to see. + if not body_failed: + for result in results: + if isinstance(result, BaseException) and not isinstance( + result, asyncio.CancelledError + ): + raise result diff --git a/fastmcp_slim/fastmcp/utilities/tests.py b/fastmcp_slim/fastmcp/utilities/tests.py index 30d354789..aad3176e9 100644 --- a/fastmcp_slim/fastmcp/utilities/tests.py +++ b/fastmcp_slim/fastmcp/utilities/tests.py @@ -1,11 +1,13 @@ from __future__ import annotations +import asyncio import copy import multiprocessing import socket import time from collections.abc import AsyncGenerator, Callable, Generator from contextlib import asynccontextmanager, contextmanager, suppress +from dataclasses import dataclass from typing import TYPE_CHECKING, Any, Literal from urllib.parse import parse_qs, urlparse @@ -15,9 +17,18 @@ from mcp.shared.auth import AuthorizationCodeResult from fastmcp import settings from fastmcp.client.auth.oauth import OAuth +from fastmcp.client.client import Client +from fastmcp.client.transports.http import StreamableHttpTransport +from fastmcp.client.transports.sse import SSETransport +from fastmcp.utilities.asgi_transport import ( + StreamingASGITransport, + run_asgi_lifespan, +) from fastmcp.utilities.http import find_available_port if TYPE_CHECKING: + from starlette.types import ASGIApp + from fastmcp.server.server import FastMCP @@ -140,6 +151,26 @@ def run_server_in_process( raise RuntimeError("Server process failed to terminate even after kill") +async def _wait_for_port(host: str, port: int, timeout: float = 5.0) -> None: + """Poll until a TCP connection to `host:port` is accepted, or raise on timeout.""" + deadline = time.monotonic() + timeout + while True: + try: + _, writer = await asyncio.open_connection(host, port) + except (ConnectionRefusedError, OSError): + if time.monotonic() >= deadline: + raise RuntimeError( + f"Server did not start listening on {host}:{port} " + f"within {timeout} seconds" + ) from None + await asyncio.sleep(0.001) + else: + writer.close() + with suppress(ConnectionResetError, BrokenPipeError): + await writer.wait_closed() + return + + @asynccontextmanager async def run_server_async( server: FastMCP, @@ -149,11 +180,13 @@ async def run_server_async( host: str = "127.0.0.1", ) -> AsyncGenerator[str, None]: """ - Start a FastMCP server as an asyncio task for in-process async testing. + Start a FastMCP server on a real port as an asyncio task. - 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. + This runs a real uvicorn server in the current process, bound to a real TCP port, + and yields its URL. Use it when the behaviour under test is genuinely about the + network — real sockets, TLS, or a server that must be reachable by something other + than an in-process client. Otherwise prefer `asgi_client` or `asgi_server`, which + exercise the same HTTP stack without binding a port. Args: server: FastMCP server instance @@ -189,14 +222,9 @@ async def run_server_async( 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( @@ -211,8 +239,9 @@ async def run_server_async( # Wait for server lifespan to be ready await server._started.wait() - # Give uvicorn a moment to bind the port after lifespan is ready - await asyncio.sleep(0.1) + # The lifespan completing does not guarantee uvicorn has bound the port yet, so + # poll until the socket accepts a connection rather than guessing at a sleep. + await _wait_for_port(host, port) try: yield f"http://{host}:{port}{path}" @@ -223,6 +252,215 @@ async def run_server_async( await asyncio.wait_for(server_task, timeout=2.0) +@dataclass(frozen=True) +class ASGIServer: + """A FastMCP server's real HTTP app, reachable in-process with no sockets. + + Yielded by `asgi_server`. The `url` looks like an ordinary server URL and the app + behind it is the genuine article — auth middleware, session manager, SSE framing and + redirects all run — but every request is dispatched straight into the ASGI + application on the current event loop. + + Because nothing is listening on the network, a plain `httpx2.AsyncClient()` cannot + reach this server. Use `client()` for a FastMCP client, `http_client()` for raw HTTP + assertions, and `transport()` when you need to build the client transport yourself. + """ + + url: str + app: ASGIApp + transport_type: Literal["http", "streamable-http", "sse"] + + def http_client( + self, + headers: dict[str, str] | None = None, + timeout: httpx2.Timeout | None = None, + auth: httpx2.Auth | None = None, + **kwargs: Any, + ) -> httpx2.AsyncClient: + """An `httpx2.AsyncClient` bound to the in-process app, for raw HTTP assertions. + + Relative URLs resolve against the server's base URL, and absolute URLs on the + same origin work too, so `client.get(f"{server.url}/health")` reads the same as + it would against a real server. + + The signature matches `McpHttpClientFactory`, so this method can also be handed + to anything that takes an `httpx_client_factory`. + """ + # The legacy SSE transport runs the whole MCP session inside its GET request and + # only releases its streams once that request observes a disconnect, so the + # bridge must let the application drain rather than cancelling at close. + cancel_on_close = self.transport_type != "sse" + return httpx2.AsyncClient( + transport=StreamingASGITransport(self.app, cancel_on_close=cancel_on_close), + base_url=self.url, + headers=headers, + timeout=timeout, + auth=auth, + **kwargs, + ) + + def transport(self, **kwargs: Any) -> StreamableHttpTransport | SSETransport: + """A FastMCP client transport wired to the in-process app. + + Accepts the same keyword arguments as the underlying transport (`headers`, + `auth`, ...); `httpx_client_factory` is supplied automatically. + """ + kwargs.setdefault("httpx_client_factory", self.http_client) + if self.transport_type == "sse": + return SSETransport(self.url, **kwargs) + return StreamableHttpTransport(self.url, **kwargs) + + def client( + self, + *, + headers: dict[str, str] | None = None, + auth: httpx2.Auth | Literal["oauth"] | str | None = None, + **client_kwargs: Any, + ) -> Client: + """An unconnected FastMCP `Client` pointed at the in-process app. + + `headers` and `auth` configure the underlying HTTP transport; every other + keyword argument is passed to `Client` (`timeout`, `elicitation_handler`, ...). + Use it as a context manager, exactly like any other client. + + Args: + headers: HTTP headers to send with every request. + auth: Client authentication, as accepted by the HTTP transports. + **client_kwargs: Additional arguments forwarded to `Client`. + """ + return Client(self.transport(headers=headers, auth=auth), **client_kwargs) + + +@asynccontextmanager +async def asgi_server( + server: FastMCP, + transport: Literal["http", "streamable-http", "sse"] = "http", + path: str | None = None, + **http_app_kwargs: Any, +) -> AsyncGenerator[ASGIServer, None]: + """ + Serve a FastMCP server's HTTP app in-process, with no socket and no uvicorn. + + This is the fastest way to test a FastMCP server over HTTP. The server's real + Starlette app is built with `http_app()` and its lifespan is started, then every + request is dispatched directly into the app on the current event loop. That skips + port binding, uvicorn startup and connection setup entirely, while still exercising + the full HTTP stack: middleware, authentication, session management and SSE + streaming all run exactly as they do in production. + + Use this as a fixture when several tests share one server but each needs its own + client. For a single test, `asgi_client` hands you a connected client in one step. + + Args: + server: FastMCP server instance. + transport: Transport type ("http", "streamable-http", or "sse"). + path: URL path for the server (defaults to "/mcp", or "/sse" for SSE). + **http_app_kwargs: Additional arguments forwarded to `server.http_app()`. + + Yields: + An `ASGIServer` describing how to reach the app. + + Example: + ```python + import pytest + from fastmcp import FastMCP + from fastmcp.utilities.tests import ASGIServer, asgi_server + + mcp = FastMCP("test") + + @mcp.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + @pytest.fixture + async def server(): + async with asgi_server(mcp) as running_server: + yield running_server + + async def test_greet(server: ASGIServer): + async with server.client() as client: + result = await client.call_tool("greet", {"name": "World"}) + assert result.data == "Hello, World!" + + async def test_greet_with_headers(server: ASGIServer): + async with server.client(headers={"X-Tenant": "acme"}) as client: + result = await client.call_tool("greet", {"name": "World"}) + assert result.data == "Hello, World!" + ``` + """ + if path is None: + path = "/sse" if transport == "sse" else "/mcp" + + app = server.http_app(transport=transport, path=path, **http_app_kwargs) + + # Nothing listens on this origin; it exists so that URLs are well-formed and so + # that host-header checks see a loopback address, as they would locally. + base_url = "http://127.0.0.1" + + async with run_asgi_lifespan(app): + yield ASGIServer( + url=f"{base_url}{path}", + app=app, + transport_type=transport, + ) + + +@asynccontextmanager +async def asgi_client( + server: FastMCP, + transport: Literal["http", "streamable-http", "sse"] = "http", + path: str | None = None, + *, + headers: dict[str, str] | None = None, + auth: httpx2.Auth | Literal["oauth"] | str | None = None, + **client_kwargs: Any, +) -> AsyncGenerator[Client, None]: + """ + Serve a FastMCP server over HTTP in-process and yield a connected `Client`. + + This is the shortest path to testing a server over a real HTTP stack. The server's + Starlette app is built and started, and requests are dispatched straight into it on + the current event loop — no port, no uvicorn, no subprocess — but middleware, + authentication, session management and SSE streaming all behave as in production. + + Reach for `asgi_server` instead when a fixture must serve several tests that each + build their own client, or when a test needs raw HTTP access to the app. + + Args: + server: FastMCP server instance. + transport: Transport type ("http", "streamable-http", or "sse"). + path: URL path for the server (defaults to "/mcp", or "/sse" for SSE). + headers: HTTP headers to send with every request. + auth: Client authentication, as accepted by the HTTP transports. + **client_kwargs: Additional arguments forwarded to `Client`. + + Yields: + A connected `Client`. + + Example: + ```python + from fastmcp import FastMCP + from fastmcp.utilities.tests import asgi_client + + async def test_greet(): + mcp = FastMCP("test") + + @mcp.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + async with asgi_client(mcp) as client: + result = await client.call_tool("greet", {"name": "World"}) + assert result.data == "Hello, World!" + ``` + """ + async with ( + asgi_server(server, transport=transport, path=path) as running_server, + running_server.client(headers=headers, auth=auth, **client_kwargs) as client, + ): + yield client + + class HeadlessOAuth(OAuth): """ OAuth provider that bypasses browser interaction for testing. diff --git a/pyproject.toml b/pyproject.toml index 91ed0fc19..b7a289254 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -126,6 +126,7 @@ env = [ markers = [ "integration: marks tests as integration tests (deselect with '-m \"not integration\"')", "client_process: marks tests that spawn client processes via stdio transport. These can create issues when run in the same CI environment as other subprocess-based tests.", + "subprocess_heavy: marks tests that spawn a fresh Python interpreter which imports FastMCP. Each one costs a full interpreter's memory and startup, so they run serially alongside client_process tests rather than competing with parallel xdist workers.", "conformance: marks MCP conformance tests (require Node.js/npx)", ] pythonpath = ["fastmcp_slim", "fastmcp_remote"] diff --git a/tests/client/client/test_session.py b/tests/client/client/test_session.py index 91ffb9a85..0579aa6d5 100644 --- a/tests/client/client/test_session.py +++ b/tests/client/client/test_session.py @@ -36,7 +36,7 @@ class TestSessionTaskErrorPropagation: async def never_complete(): """A coroutine that will never complete normally.""" - await asyncio.sleep(1000) + await asyncio.Event().wait() async def failing_session(): """Simulates a session task that raises an error.""" diff --git a/tests/client/minimal_stdio_server.py b/tests/client/minimal_stdio_server.py new file mode 100644 index 000000000..abca59f76 --- /dev/null +++ b/tests/client/minimal_stdio_server.py @@ -0,0 +1,192 @@ +"""A minimal MCP server spoken over stdio, using only the standard library. + +This is a **test fixture**, not a real server. It exists so that subprocess +lifecycle tests (keep-alive, crash recovery, PID identity) can spawn many +short-lived servers without paying for `import fastmcp` in every child +process. Importing fastmcp and constructing a `FastMCP` instance costs +roughly 0.7s per spawn; this script starts in roughly 0.03s. + +It implements only what those tests exercise: the `initialize` handshake, +`tools/list`, and `tools/call` for two trivial tools. Anything that needs +real FastMCP semantics (tool serialization, error handling, structured +output shapes) must use a real FastMCP server instead. + +The response shapes below were captured from the wire of a real FastMCP +stdio server so that `CallToolResult.data` deserializes identically. + +Usage: + + python minimal_stdio_server.py [--exit-after-calls N] + +With `--exit-after-calls N`, the `pid` tool schedules a clean `os._exit(0)` +shortly after its Nth invocation, simulating a server that shuts itself +down mid-session. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +import threading +from typing import Any + +INT_OUTPUT_SCHEMA: dict[str, Any] = { + "properties": {"result": {"type": "integer"}}, + "required": ["result"], + "type": "object", + "x-fastmcp-wrap-result": True, +} + +STR_OUTPUT_SCHEMA: dict[str, Any] = { + "properties": {"result": {"type": "string"}}, + "required": ["result"], + "type": "object", + "x-fastmcp-wrap-result": True, +} + +TOOLS: list[dict[str, Any]] = [ + { + "name": "pid", + "description": "Gets PID of server", + "inputSchema": { + "properties": {}, + "type": "object", + "additionalProperties": False, + }, + "outputSchema": INT_OUTPUT_SCHEMA, + }, + { + "name": "echo", + "description": "Echoes the message back", + "inputSchema": { + "properties": {"message": {"type": "string"}}, + "required": ["message"], + "type": "object", + "additionalProperties": False, + }, + "outputSchema": STR_OUTPUT_SCHEMA, + }, +] + +METHOD_NOT_FOUND = -32601 +INVALID_PARAMS = -32602 + + +def _wrapped_result(value: int | str) -> dict[str, Any]: + """Mirror how FastMCP reports a scalar return value on the wire.""" + return { + "_meta": {"fastmcp": {"wrap_result": True}}, + "content": [{"type": "text", "text": str(value)}], + "isError": False, + "structuredContent": {"result": value}, + } + + +class MinimalServer: + def __init__(self, exit_after_calls: int | None) -> None: + self.exit_after_calls = exit_after_calls + self.pid_call_count = 0 + + def send(self, message: dict[str, Any]) -> None: + sys.stdout.write(json.dumps(message) + "\n") + sys.stdout.flush() + + def reply(self, request_id: Any, result: dict[str, Any]) -> None: + self.send({"jsonrpc": "2.0", "id": request_id, "result": result}) + + def reply_error(self, request_id: Any, code: int, message: str) -> None: + self.send( + { + "jsonrpc": "2.0", + "id": request_id, + "error": {"code": code, "message": message}, + } + ) + + def handle_initialize(self, request_id: Any, params: dict[str, Any]) -> None: + # Echo the client's requested version back. The client rejects any + # version it did not ask for, and echoing keeps this fixture working + # across SDK protocol bumps without edits. + protocol_version = params.get("protocolVersion") + self.reply( + request_id, + { + "protocolVersion": protocol_version, + "capabilities": {"tools": {"listChanged": False}}, + "serverInfo": {"name": "MinimalStdioServer", "version": "1.0.0"}, + }, + ) + + def handle_tools_call(self, request_id: Any, params: dict[str, Any]) -> None: + name = params.get("name") + arguments = params.get("arguments") or {} + + if name == "pid": + self.pid_call_count += 1 + pid = os.getpid() + if ( + self.exit_after_calls is not None + and self.pid_call_count >= self.exit_after_calls + ): + # Reply first, then exit shortly after, so the client sees a + # successful call followed by an unannounced clean shutdown. + self.reply(request_id, _wrapped_result(pid)) + threading.Timer(0.1, lambda: os._exit(0)).start() + return + self.reply(request_id, _wrapped_result(pid)) + return + + if name == "echo": + message = arguments.get("message") + if not isinstance(message, str): + self.reply_error(request_id, INVALID_PARAMS, "message must be a string") + return + self.reply(request_id, _wrapped_result(message)) + return + + self.reply_error(request_id, INVALID_PARAMS, f"Unknown tool: {name}") + + def handle(self, message: dict[str, Any]) -> None: + method = message.get("method") + request_id = message.get("id") + params = message.get("params") or {} + + if request_id is None: + # Notification (e.g. notifications/initialized) — nothing to send. + return + + if method == "initialize": + self.handle_initialize(request_id, params) + elif method == "ping": + self.reply(request_id, {}) + elif method == "tools/list": + self.reply(request_id, {"tools": TOOLS}) + elif method == "tools/call": + self.handle_tools_call(request_id, params) + else: + self.reply_error(request_id, METHOD_NOT_FOUND, f"Unknown method: {method}") + + def run(self) -> None: + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + message = json.loads(line) + except json.JSONDecodeError: + continue + if isinstance(message, dict): + self.handle(message) + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--exit-after-calls", type=int, default=None) + parsed = parser.parse_args() + MinimalServer(exit_after_calls=parsed.exit_after_calls).run() + + +if __name__ == "__main__": + main() diff --git a/tests/client/tasks/test_client_task_notifications.py b/tests/client/tasks/test_client_task_notifications.py index 763622fa2..1628eb151 100644 --- a/tests/client/tasks/test_client_task_notifications.py +++ b/tests/client/tasks/test_client_task_notifications.py @@ -7,6 +7,7 @@ and invoke user callbacks. import asyncio import time +from collections.abc import Callable from datetime import datetime, timezone import pytest @@ -16,6 +17,17 @@ from fastmcp import FastMCP from fastmcp.client import Client +async def _wait_until(condition: Callable[[], bool], timeout: float = 5.0) -> None: + """Poll until condition() is true or timeout elapses. + + Used in place of a fixed sleep when waiting for an async callback or + notification to be delivered/dispatched after the awaited call returns. + """ + deadline = time.monotonic() + timeout + while not condition() and time.monotonic() < deadline: + await asyncio.sleep(0.005) + + @pytest.fixture async def task_notification_server(): """Server that sends task status notifications.""" @@ -23,8 +35,8 @@ async def task_notification_server(): @mcp.tool(task=True) async def quick_task(value: int) -> int: - """Quick background task.""" - await asyncio.sleep(0.05) + """Quick background task with a brief, measurable delay (contrast with instant_task).""" + await asyncio.sleep(0.01) return value * 2 @mcp.tool(task=True) @@ -32,12 +44,6 @@ async def task_notification_server(): """Background task that completes with no delay.""" return value * 2 - @mcp.tool(task=True) - async def slow_task(duration: float = 0.2) -> str: - """Slow background task.""" - await asyncio.sleep(duration) - return "done" - @mcp.tool(task=True) async def failing_task() -> str: """Task that fails.""" @@ -93,8 +99,12 @@ async def test_callback_invoked_on_notification(task_notification_server): # Wait for completion await task.wait(timeout=2.0) - # Give callbacks a moment to fire - await asyncio.sleep(0.1) + # Wait for the status this test actually asserts on. Waiting merely for + # "some callback fired" would be satisfied by the earlier `working` + # notification and race the `completed` one. + await _wait_until( + lambda: any(s.status == "completed" for s in callback_invocations) + ) # Callback should have been invoked at least once assert len(callback_invocations) > 0 @@ -123,7 +133,7 @@ async def test_async_callback_invoked(task_notification_server): await task.wait(timeout=2.0) # Give async callbacks time to complete - await asyncio.sleep(0.2) + await _wait_until(lambda: len(callback_invocations) > 0) # Async callback should have been invoked assert len(callback_invocations) > 0 @@ -147,7 +157,7 @@ async def test_multiple_callbacks_all_invoked(task_notification_server): task.on_status_change(callback2) await task.wait(timeout=2.0) - await asyncio.sleep(0.1) + await _wait_until(lambda: bool(callback1_calls) and bool(callback2_calls)) # Both callbacks should have been invoked assert len(callback1_calls) > 0 @@ -173,7 +183,7 @@ async def test_callback_error_doesnt_break_notification(task_notification_server task.on_status_change(working_callback) await task.wait(timeout=2.0) - await asyncio.sleep(0.1) + await _wait_until(lambda: bool(callback1_calls) and bool(callback2_calls)) # Failing callback was called (and errored) assert len(callback1_calls) > 0 @@ -240,7 +250,7 @@ async def test_fast_task_completion_delivered_via_notification( assert result.data == 42 # Allow the completion notification to arrive and dispatch. - await asyncio.sleep(0.1) + await _wait_until(lambda: "completed" in received) assert "completed" in received diff --git a/tests/client/telemetry/test_client_task_tracing.py b/tests/client/telemetry/test_client_task_tracing.py index 4d0b71718..f910b6cc8 100644 --- a/tests/client/telemetry/test_client_task_tracing.py +++ b/tests/client/telemetry/test_client_task_tracing.py @@ -71,7 +71,9 @@ async def test_task_id_operations_create_propagating_client_spans( @server.tool(task=True) async def slow_tool() -> str: started.set() - await asyncio.sleep(10) + # Never completes on its own - the test cancels this task well + # before any real-time completion would matter. + await asyncio.Event().wait() return "done" async with Client(server) as client: diff --git a/tests/client/test_oauth_callback_race.py b/tests/client/test_oauth_callback_race.py index 83622a86a..417173ea9 100644 --- a/tests/client/test_oauth_callback_race.py +++ b/tests/client/test_oauth_callback_race.py @@ -8,6 +8,18 @@ from fastmcp.client.oauth_callback import ( from fastmcp.utilities.http import find_available_port +async def _wait_until_listening(server) -> None: + """Poll until the callback server's socket is accepting connections. + + uvicorn sets `Server.started = True` right after it binds and starts + listening on the socket, before `serve()` moves on to request handling, + so this is a deterministic readiness signal in place of a fixed sleep. + """ + with anyio.fail_after(5): + while not server.started: + await anyio.sleep(0.001) + + async def test_oauth_callback_result_ignores_subsequent_callbacks(): """Only the first callback should be captured in shared OAuth callback state.""" port = find_available_port() @@ -22,7 +34,7 @@ async def test_oauth_callback_result_ignores_subsequent_callbacks(): async with anyio.create_task_group() as tg: tg.start_soon(server.serve) - await anyio.sleep(0.05) + await _wait_until_listening(server) async with httpx2.AsyncClient() as client: first = await client.get( @@ -72,7 +84,7 @@ async def test_oauth_callback_result_captures_iss(): async with anyio.create_task_group() as tg: tg.start_soon(server.serve) - await anyio.sleep(0.05) + await _wait_until_listening(server) async with httpx2.AsyncClient() as client: response = await client.get( @@ -112,7 +124,7 @@ async def test_oauth_callback_result_captures_iss_on_error(): async with anyio.create_task_group() as tg: tg.start_soon(server.serve) - await anyio.sleep(0.05) + await _wait_until_listening(server) async with httpx2.AsyncClient() as client: response = await client.get( diff --git a/tests/client/test_sampling_tool_loop.py b/tests/client/test_sampling_tool_loop.py index 809e58ed8..0e18bd25f 100644 --- a/tests/client/test_sampling_tool_loop.py +++ b/tests/client/test_sampling_tool_loop.py @@ -285,26 +285,27 @@ class TestAutomaticToolLoop: async def test_concurrent_tool_execution_default_sequential(self): """Test that tools execute sequentially by default.""" import asyncio - import time from mcp_types import CreateMessageResultWithTools, ToolUseContent - execution_order: list[tuple[str, float]] = [] + # Ordering is guaranteed structurally (the loop awaits each tool call + # to completion before starting the next when tool_concurrency is + # None), so no real delay is needed to prove it - a single + # `asyncio.sleep(0)` still yields control to the event loop. + execution_order: list[str] = [] async def slow_tool_a(x: int) -> int: """Slow tool A.""" - start = time.time() - execution_order.append(("tool_a_start", start)) - await asyncio.sleep(0.1) - execution_order.append(("tool_a_end", time.time())) + execution_order.append("tool_a_start") + await asyncio.sleep(0) + execution_order.append("tool_a_end") return x * 2 async def slow_tool_b(y: int) -> int: """Slow tool B.""" - start = time.time() - execution_order.append(("tool_b_start", start)) - await asyncio.sleep(0.1) - execution_order.append(("tool_b_end", time.time())) + execution_order.append("tool_b_start") + await asyncio.sleep(0) + execution_order.append("tool_b_end") return y + 10 call_count = 0 @@ -359,30 +360,39 @@ class TestAutomaticToolLoop: assert result.data == "Done!" # Verify sequential execution: tool_a must complete before tool_b starts - events = [e[0] for e in execution_order] - assert events == ["tool_a_start", "tool_a_end", "tool_b_start", "tool_b_end"] + assert execution_order == [ + "tool_a_start", + "tool_a_end", + "tool_b_start", + "tool_b_end", + ] async def test_concurrent_tool_execution_unlimited(self): """Test unlimited parallel tool execution with tool_concurrency=0.""" import asyncio - import time from mcp_types import CreateMessageResultWithTools, ToolUseContent - execution_times: dict[str, dict[str, float]] = {} + # tool_a blocks on an event that only tool_b sets. This is only + # satisfiable if both tools are genuinely running concurrently: under + # sequential execution tool_b would never start (tool_a would never + # finish awaiting it) and the test would fail via the wait_for + # timeout rather than racing on wall-clock timestamps. + execution_order: list[str] = [] + tool_b_started = asyncio.Event() async def slow_tool_a(x: int) -> int: """Slow tool A.""" - execution_times["tool_a"] = {"start": time.time()} - await asyncio.sleep(0.1) - execution_times["tool_a"]["end"] = time.time() + execution_order.append("tool_a_start") + await asyncio.wait_for(tool_b_started.wait(), timeout=1.0) + execution_order.append("tool_a_end") return x * 2 async def slow_tool_b(y: int) -> int: """Slow tool B.""" - execution_times["tool_b"] = {"start": time.time()} - await asyncio.sleep(0.1) - execution_times["tool_b"]["end"] = time.time() + execution_order.append("tool_b_start") + tool_b_started.set() + execution_order.append("tool_b_end") return y + 10 call_count = 0 @@ -436,26 +446,41 @@ class TestAutomaticToolLoop: result = await client.call_tool("test_tool", {}) assert result.data == "Done!" - # Verify parallel execution: both tools should overlap in time - assert "tool_a" in execution_times - assert "tool_b" in execution_times - # tool_b should start before tool_a finishes (overlap) - assert execution_times["tool_b"]["start"] < execution_times["tool_a"]["end"] + # Verify parallel execution: tool_b started and finished entirely + # inside tool_a's blocked wait, which is only possible if the two + # tools were running concurrently. + assert execution_order == [ + "tool_a_start", + "tool_b_start", + "tool_b_end", + "tool_a_end", + ] async def test_concurrent_tool_execution_bounded(self): """Test bounded parallel execution with tool_concurrency=2.""" import asyncio - import time from mcp_types import CreateMessageResultWithTools, ToolUseContent - execution_order: list[tuple[str, float]] = [] + # tool_1 and tool_2 each block until *both* have started, which is + # only possible if two slots are occupied simultaneously (proving + # concurrency=2 admits two tools at once). tool_3 has no blocking + # branch, so its appearance in the log tells us when the real + # semaphore in the implementation let it in - only after a slot + # frees, i.e. after tool_1 or tool_2 finishes. + execution_order: list[str] = [] + both_started = asyncio.Event() + started_names: set[str] = set() - async def slow_tool(name: str, duration: float = 0.1) -> str: - """Generic slow tool.""" - execution_order.append((f"{name}_start", time.time())) - await asyncio.sleep(duration) - execution_order.append((f"{name}_end", time.time())) + async def slow_tool(name: str) -> str: + """Generic tool used to observe bounded concurrency.""" + execution_order.append(f"{name}_start") + if name in ("tool_1", "tool_2"): + started_names.add(name) + if {"tool_1", "tool_2"} <= started_names: + both_started.set() + await asyncio.wait_for(both_started.wait(), timeout=1.0) + execution_order.append(f"{name}_end") return f"{name} done" call_count = 0 @@ -475,19 +500,19 @@ class TestAutomaticToolLoop: type="tool_use", id="call_1", name="slow_tool", - input={"name": "tool_1", "duration": 0.1}, + input={"name": "tool_1"}, ), ToolUseContent( type="tool_use", id="call_2", name="slow_tool", - input={"name": "tool_2", "duration": 0.1}, + input={"name": "tool_2"}, ), ToolUseContent( type="tool_use", id="call_3", name="slow_tool", - input={"name": "tool_3", "duration": 0.05}, + input={"name": "tool_3"}, ), ], model="test-model", @@ -517,38 +542,39 @@ class TestAutomaticToolLoop: assert result.data == "Done!" # Verify that at most 2 tools run concurrently - events = [e[0] for e in execution_order] # First 2 tools should start before either ends - assert events[0] in ["tool_1_start", "tool_2_start"] - assert events[1] in ["tool_1_start", "tool_2_start"] + assert execution_order[0] in ["tool_1_start", "tool_2_start"] + assert execution_order[1] in ["tool_1_start", "tool_2_start"] # Third tool should start after at least one of the first two finishes - tool_3_start_idx = events.index("tool_3_start") + tool_3_start_idx = execution_order.index("tool_3_start") assert ( - "tool_1_end" in events[:tool_3_start_idx] - or "tool_2_end" in events[:tool_3_start_idx] + "tool_1_end" in execution_order[:tool_3_start_idx] + or "tool_2_end" in execution_order[:tool_3_start_idx] ) async def test_sequential_tool_forces_sequential_execution(self): """Test that sequential=True forces all tools to execute sequentially.""" import asyncio - import time from mcp_types import CreateMessageResultWithTools, ToolUseContent - execution_order: list[tuple[str, float]] = [] + # A sequential=True tool in the batch forces the whole batch through + # the plain for-loop path (see run.py's `requires_sequential`), so + # ordering is guaranteed structurally and no real delay is needed. + execution_order: list[str] = [] async def normal_tool(x: int) -> int: """Normal tool.""" - execution_order.append(("normal_start", time.time())) - await asyncio.sleep(0.05) - execution_order.append(("normal_end", time.time())) + execution_order.append("normal_start") + await asyncio.sleep(0) + execution_order.append("normal_end") return x * 2 async def sequential_tool(y: int) -> int: """Sequential tool.""" - execution_order.append(("sequential_start", time.time())) - await asyncio.sleep(0.05) - execution_order.append(("sequential_end", time.time())) + execution_order.append("sequential_start") + await asyncio.sleep(0) + execution_order.append("sequential_end") return y + 10 call_count = 0 @@ -607,16 +633,15 @@ class TestAutomaticToolLoop: assert result.data == "Done!" # Verify sequential execution: first tool must complete before second starts - events = [e[0] for e in execution_order] - assert events[0] in ["normal_start", "sequential_start"] - assert events[1] in ["normal_end", "sequential_end"] + assert execution_order[0] in ["normal_start", "sequential_start"] + assert execution_order[1] in ["normal_end", "sequential_end"] # Ensure the second tool starts after the first ends - if events[0] == "normal_start": - assert events[1] == "normal_end" - assert events[2] == "sequential_start" + if execution_order[0] == "normal_start": + assert execution_order[1] == "normal_end" + assert execution_order[2] == "sequential_start" else: - assert events[1] == "sequential_end" - assert events[2] == "normal_start" + assert execution_order[1] == "sequential_end" + assert execution_order[2] == "normal_start" async def test_concurrent_tool_execution_error_handling(self): """Test that errors are captured per-tool in parallel execution.""" @@ -695,9 +720,26 @@ class TestAutomaticToolLoop: ToolUseContent, ) - async def tool_with_delay(value: int, delay: float) -> int: - """Tool that takes variable time.""" - await asyncio.sleep(delay) + # Chain events so the tools finish in a different order (2, 3, 1) + # than they were called (1, 2, 3), without depending on real delays: + # tool 2 finishes immediately and unblocks tool 3, which finishes and + # unblocks tool 1. This only resolves if all three run concurrently - + # under sequential execution tool 1 would deadlock waiting on tool 3, + # which itself would never have been started yet. + tool_2_done = asyncio.Event() + tool_3_done = asyncio.Event() + + async def tool_with_delay(value: int) -> int: + """Tool that finishes out of call order.""" + if value == 1: + await asyncio.wait_for(tool_3_done.wait(), timeout=1.0) + elif value == 3: + await asyncio.wait_for(tool_2_done.wait(), timeout=1.0) + + if value == 2: + tool_2_done.set() + elif value == 3: + tool_3_done.set() return value messages_received: list[list[SamplingMessage]] = [] @@ -708,7 +750,7 @@ class TestAutomaticToolLoop: messages_received.append(list(messages)) if len(messages_received) == 1: - # Tools with different delays - later tools finish first + # Call order is 1, 2, 3 but they finish out of order (2, 3, 1) return CreateMessageResultWithTools( role="assistant", content=[ @@ -716,19 +758,19 @@ class TestAutomaticToolLoop: type="tool_use", id="call_1", name="tool_with_delay", - input={"value": 1, "delay": 0.15}, + input={"value": 1}, ), ToolUseContent( type="tool_use", id="call_2", name="tool_with_delay", - input={"value": 2, "delay": 0.05}, + input={"value": 2}, ), ToolUseContent( type="tool_use", id="call_3", name="tool_with_delay", - input={"value": 3, "delay": 0.1}, + input={"value": 3}, ), ], model="test-model", diff --git a/tests/client/test_slim_package_boundaries.py b/tests/client/test_slim_package_boundaries.py index 5f59b6b3c..187bf177a 100644 --- a/tests/client/test_slim_package_boundaries.py +++ b/tests/client/test_slim_package_boundaries.py @@ -64,6 +64,7 @@ async def test_multiserver_config_requires_server_for_now() -> None: pass +@pytest.mark.subprocess_heavy def test_bare_slim_import_needs_only_mcp_types() -> None: """A bare `fastmcp-slim` install ships `mcp-types` but not the full `mcp` SDK. diff --git a/tests/client/test_sse.py b/tests/client/test_sse.py index eb94faa75..15a8bda34 100644 --- a/tests/client/test_sse.py +++ b/tests/client/test_sse.py @@ -11,7 +11,7 @@ from fastmcp.client.transports import SSETransport from fastmcp.server.dependencies import get_http_request from fastmcp.server.http import create_sse_app from fastmcp.server.server import FastMCP -from fastmcp.utilities.tests import run_server_async +from fastmcp.utilities.tests import ASGIServer, asgi_server def create_test_server() -> FastMCP: @@ -63,24 +63,22 @@ def create_test_server() -> FastMCP: @pytest.fixture async def sse_server(): - """Start a test server with SSE transport and return its URL.""" + """Start a test server with SSE transport, in-process.""" server = create_test_server() - async with run_server_async(server, transport="sse") as url: - yield url + async with asgi_server(server, transport="sse") as running_server: + yield running_server -async def test_ping(sse_server: str): +async def test_ping(sse_server: ASGIServer): """Test pinging the server.""" - async with Client(transport=SSETransport(sse_server)) as client: + async with sse_server.client() as client: result = await client.ping() assert result is True -async def test_http_headers(sse_server: str): +async def test_http_headers(sse_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"}) - ) as client: + async with sse_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: raw_result = await client.read_resource("request://headers") assert isinstance(raw_result[0], TextResourceContents) json_result = json.loads(raw_result[0].text) @@ -90,10 +88,10 @@ async def test_http_headers(sse_server: str): @pytest.fixture async def sse_server_custom_path(): - """Start a test server with SSE on a custom path.""" + """Start a test server with SSE on a custom path, in-process.""" server = create_test_server() - async with run_server_async(server, transport="sse", path="/help") as url: - yield url + async with asgi_server(server, transport="sse", path="/help") as running_server: + yield running_server @pytest.fixture @@ -140,9 +138,9 @@ async def nested_sse_server(): pass -async def test_run_server_on_path(sse_server_custom_path: str): +async def test_run_server_on_path(sse_server_custom_path: ASGIServer): """Test running server on a custom path.""" - async with Client(transport=SSETransport(sse_server_custom_path)) as client: + async with sse_server_custom_path.client() as client: result = await client.ping() assert result is True @@ -159,42 +157,33 @@ async def test_nested_sse_server_resolves_correctly(nested_sse_server: str): reason="Timeout tests are flaky on Windows. Timeouts *are* supported but the tests are unreliable.", ) class TestTimeout: - async def test_timeout(self, sse_server: str): + async def test_timeout(self, sse_server: ASGIServer): with pytest.raises( MCPError, match="timed out", ): - async with Client( - transport=SSETransport(sse_server), - timeout=0.03, - ) as client: + async with sse_server.client(timeout=0.03) as client: await client.call_tool("sleep", {"seconds": 0.1}) - async def test_timeout_tool_call(self, sse_server: str): - async with Client(transport=SSETransport(sse_server)) as client: + async def test_timeout_tool_call(self, sse_server: ASGIServer): + async with sse_server.client() as client: with pytest.raises(MCPError, match="timed out"): await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.03) async def test_timeout_tool_call_overrides_client_timeout_if_lower( - self, sse_server: str + self, sse_server: ASGIServer ): - async with Client( - transport=SSETransport(sse_server), - timeout=2, - ) as client: + async with sse_server.client(timeout=2) as client: with pytest.raises(MCPError, match="timed out"): await client.call_tool("sleep", {"seconds": 0.1}, timeout=0.03) async def test_timeout_client_timeout_does_not_override_tool_call_timeout_if_lower( - self, sse_server: str + self, sse_server: ASGIServer ): """ With SSE, the tool call timeout always takes precedence over the client. Note: on Windows, the behavior appears unpredictable. """ - async with Client( - transport=SSETransport(sse_server), - timeout=0.5, - ) as client: + async with sse_server.client(timeout=0.5) as client: await client.call_tool("sleep", {"seconds": 0.8}, timeout=2) diff --git a/tests/client/test_stdio.py b/tests/client/test_stdio.py index af4995203..c6eec09b8 100644 --- a/tests/client/test_stdio.py +++ b/tests/client/test_stdio.py @@ -2,14 +2,23 @@ import asyncio import gc import inspect import os +import time import weakref +from pathlib import Path import psutil import pytest +from mcp.shared.exceptions import MCPError from fastmcp import Client from fastmcp.client.transports import PythonStdioTransport, StdioTransport +# A pure-stdlib MCP server used by the process-lifecycle tests below. It starts +# in ~0.03s instead of the ~0.7s a real FastMCP server needs, which matters +# because these tests spawn several subprocesses each. See its docstring for +# what it does and does not implement. +MINIMAL_STDIO_SERVER = Path(__file__).parent / "minimal_stdio_server.py" + def running_under_debugger(): return os.environ.get("DEBUGPY_RUNNING") == "true" @@ -24,26 +33,48 @@ def gc_collect_harder(): gc.collect() +async def wait_for_log_content( + log_file_path, expected: str, timeout: float = 2.0 +) -> str: + """Poll a log file until it contains the expected text. + + The subprocess's stderr is redirected straight to the file at the OS + level (no async pump on our side to synchronize on), so poll for the + content instead of sleeping a fixed amount and hoping it landed. + """ + + async def _poll() -> str: + while True: + content = log_file_path.read_text() + if expected in content: + return content + await asyncio.sleep(0.01) + + return await asyncio.wait_for(_poll(), timeout=timeout) + + +async def wait_for_process_exit(pid: int | None, timeout: float = 5.0) -> None: + """Poll until the given pid is gone, failing clearly if it never exits. + + The subprocesses under test self-terminate within a fraction of a second, + so a bounded poll costs nothing and turns a hung teardown into a named + failure instead of an opaque suite-level timeout. + """ + assert pid is not None + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + psutil.Process(pid) + except psutil.NoSuchProcess: + return + await asyncio.sleep(0.01) + pytest.fail(f"Subprocess {pid} was still alive after {timeout}s") + + class TestParallelCalls: @pytest.fixture - def stdio_script(self, tmp_path): - script = inspect.cleandoc(''' - import os - from fastmcp import FastMCP - - mcp = FastMCP() - - @mcp.tool - def pid() -> int: - """Gets PID of server""" - return os.getpid() - - if __name__ == "__main__": - mcp.run() - ''') - script_file = tmp_path / "stdio.py" - script_file.write_text(script) - return script_file + def stdio_script(self): + return MINIMAL_STDIO_SERVER async def test_parallel_calls(self, stdio_script): from fastmcp.server import create_proxy @@ -69,24 +100,8 @@ class TestKeepAlive: # https://github.com/PrefectHQ/fastmcp/issues/581 @pytest.fixture - def stdio_script(self, tmp_path): - script = inspect.cleandoc(''' - import os - from fastmcp import FastMCP - - mcp = FastMCP() - - @mcp.tool - def pid() -> int: - """Gets PID of server""" - return os.getpid() - - if __name__ == "__main__": - mcp.run() - ''') - script_file = tmp_path / "stdio.py" - script_file.write_text(script) - return script_file + def stdio_script(self): + return MINIMAL_STDIO_SERVER async def test_keep_alive_default_true(self): client = Client(transport=StdioTransport(command="python", args=[""])) @@ -160,10 +175,7 @@ class TestKeepAlive: # This test may fail/hang while debugging because the debugger holds a reference to the underlying transport - with pytest.raises(psutil.NoSuchProcess): - while True: - psutil.Process(pid) - await asyncio.sleep(0.1) + await wait_for_process_exit(pid) async def test_keep_alive_false_exit_scope_kills_server(self, stdio_script): pid: int | None = None @@ -181,10 +193,7 @@ class TestKeepAlive: await test_server() - with pytest.raises(psutil.NoSuchProcess): - while True: - psutil.Process(pid) - await asyncio.sleep(0.1) + await wait_for_process_exit(pid) async def test_keep_alive_false_starts_new_session_across_multiple_calls( self, stdio_script @@ -268,24 +277,8 @@ class TestSubprocessCrashRecovery: INIT_TIMEOUT = 3 @pytest.fixture - def stdio_script(self, tmp_path): - script = inspect.cleandoc(''' - import os - from fastmcp import FastMCP - - mcp = FastMCP() - - @mcp.tool - def pid() -> int: - """Gets PID of server""" - return os.getpid() - - if __name__ == "__main__": - mcp.run() - ''') - script_file = tmp_path / "stdio.py" - script_file.write_text(script) - return script_file + def stdio_script(self): + return MINIMAL_STDIO_SERVER async def test_keep_alive_recovers_after_subprocess_crash(self, stdio_script): """When keep_alive=True and the subprocess dies, the next connection should start a fresh subprocess.""" @@ -431,61 +424,77 @@ class TestSubprocessCrashRecovery: # Kill the subprocess psutil.Process(pid1).kill() - # Fire several concurrent requests — all should fail, none should hang + # Fire several concurrent requests. Depending on how quickly the dead + # session is detected and a replacement spawned, each caller either + # fails cleanly or lands on the fresh subprocess — with a fast-starting + # server, recovery can beat all five requests and the crash is fully + # transparent. What must never happen: a hang (gather returning is the + # proof), or a "success" served by the killed process. tasks = [proxy.call_tool("pid") for _ in range(5)] results = await asyncio.gather(*tasks, return_exceptions=True) - errors = [r for r in results if isinstance(r, Exception)] - assert len(errors) > 0 + for r in results: + if not isinstance(r, Exception): + served_by = int(r.content[0].text) # type: ignore[union-attr] # ty:ignore[unresolved-attribute] + assert served_by != pid1, ( + "call reported success from the killed subprocess" + ) - # Recovery: a subsequent request should succeed + # Recovery: a subsequent request must succeed on a fresh subprocess result = await proxy.call_tool("pid") pid2 = int(result.content[0].text) # type: ignore[union-attr] # ty:ignore[unresolved-attribute] assert pid1 != pid2 - async def test_clean_exit_recovers(self, tmp_path): + async def test_clean_exit_recovers(self): """Recovery works when the subprocess exits cleanly (exit code 0), not just crashes.""" - script = tmp_path / "exit_script.py" - script.write_text( - inspect.cleandoc(''' - import os, sys, threading - from fastmcp import FastMCP - - mcp = FastMCP() - call_count = 0 - - @mcp.tool - def pid_then_exit() -> int: - """Returns PID, exits cleanly after second call.""" - global call_count - call_count += 1 - pid = os.getpid() - if call_count >= 2: - threading.Timer(0.1, lambda: os._exit(0)).start() - return pid - - if __name__ == "__main__": - mcp.run() - ''') - ) - client = Client( - transport=PythonStdioTransport(script_path=script), + transport=PythonStdioTransport( + script_path=MINIMAL_STDIO_SERVER, + args=["--exit-after-calls", "2"], + ), init_timeout=self.INIT_TIMEOUT, ) async with client: - result1 = await client.call_tool("pid_then_exit") + result1 = await client.call_tool("pid") pid1: int = result1.data # Second call triggers delayed clean exit - await client.call_tool("pid_then_exit") - await asyncio.sleep(0.3) + await client.call_tool("pid") - # Recovery after clean exit - async with client: - result2 = await client.call_tool("pid_then_exit") - pid2: int = result2.data + # Wait for the subprocess to actually exit (it self-terminates + # via a background timer ~0.1s after the second call) instead + # of blindly sleeping past the worst case. + await wait_for_process_exit(pid1) + # Recovery after clean exit. + # + # The transport only notices a dead session once the SDK dispatcher's + # read loop has observed EOF on the subprocess's stdout and set its + # `_closed` flag (see `StdioTransport._is_session_dead`). The process + # being gone does not imply that detection has happened yet: EOF has to + # travel from the OS pipe through anyio's stream plumbing and then be + # picked up by a separate read-loop task. On a loaded machine — notably + # Windows CI running xdist workers on two cores — that can land after + # `connect()` samples the flag, so the first attempt is routed to the + # stale session and fails with CONNECTION_CLOSED, which in turn tears + # the session down so the next attempt reconnects. + # + # The crash tests above encode this same one-failure-then-recover + # contract explicitly with `pytest.raises`. Here the failure is + # timing-dependent rather than guaranteed, so retry once instead: the + # invariant under test is that a cleanly-exited server is replaced by a + # fresh subprocess, not how many attempts EOF detection costs. + pid2: int | None = None + for _ in range(2): + try: + async with client: + result2 = await client.call_tool("pid") + pid2 = result2.data + break + except (MCPError, RuntimeError): + continue + + assert pid2 is not None, "Client did not recover after a clean subprocess exit" assert pid1 != pid2 async def test_crash_during_initialization(self, tmp_path): @@ -508,20 +517,13 @@ class TestSubprocessCrashRecovery: async with client: pass - # Write a working script to the same path + # Replace the same path with a working server. It delegates to the + # minimal stdio server so the retry doesn't pay for a fastmcp import. crash_script.write_text( - inspect.cleandoc(""" - import os - from fastmcp import FastMCP + inspect.cleandoc(f""" + import runpy - mcp = FastMCP() - - @mcp.tool - def pid() -> int: - return os.getpid() - - if __name__ == "__main__": - mcp.run() + runpy.run_path({str(MINIMAL_STDIO_SERVER)!r}, run_name="__main__") """) ) @@ -531,7 +533,16 @@ class TestSubprocessCrashRecovery: assert isinstance(result.data, int) +@pytest.mark.subprocess_heavy class TestLogFile: + """Stderr capture, proven against a real FastMCP server. + + Unlike the rest of this module these spawn a full `import fastmcp` + interpreter rather than the minimal stdlib server, because the point is + that the log file captures a real server's stderr. That costs ~0.7s per + spawn, so they run in the serial CI step. + """ + @pytest.fixture def stdio_script_with_stderr(self, tmp_path): script = inspect.cleandoc(''' @@ -594,10 +605,7 @@ class TestLogFile: async with client: await client.call_tool("write_error", {"message": "Test error message"}) - # Need to wait a bit for stderr to flush - await asyncio.sleep(0.1) - - content = log_file_path.read_text() + content = await wait_for_log_content(log_file_path, "Test error message") assert "Test error message" in content async def test_log_file_captures_stderr_output_with_textio( @@ -617,10 +625,10 @@ class TestLogFile: "write_error", {"message": "Test error with TextIO"} ) - # Need to wait a bit for stderr to flush - await asyncio.sleep(0.1) + content = await wait_for_log_content( + log_file_path, "Test error with TextIO" + ) - content = log_file_path.read_text() assert "Test error with TextIO" in content async def test_log_file_none_uses_default_behavior( diff --git a/tests/client/test_streamable_http.py b/tests/client/test_streamable_http.py index f109587e6..e50fcfdba 100644 --- a/tests/client/test_streamable_http.py +++ b/tests/client/test_streamable_http.py @@ -13,7 +13,7 @@ from fastmcp.client import Client from fastmcp.client.transports import StreamableHttpTransport from fastmcp.server.dependencies import get_http_request from fastmcp.server.server import FastMCP -from fastmcp.utilities.tests import run_server_async +from fastmcp.utilities.tests import ASGIServer, asgi_server def create_test_server() -> FastMCP: @@ -90,8 +90,8 @@ async def streamable_http_server(request): fastmcp.settings.stateless_http = True server = create_test_server() - async with run_server_async(server) as url: - yield url + async with asgi_server(server) as running_server: + yield running_server if stateless_http: fastmcp.settings.stateless_http = False @@ -101,8 +101,8 @@ async def streamable_http_server(request): async def streamable_http_server_with_streamable_http_alias(): """Test that the "streamable-http" transport alias works.""" server = create_test_server() - async with run_server_async(server, transport="streamable-http") as url: - yield url + async with asgi_server(server, transport="streamable-http") as running_server: + yield running_server @pytest.fixture @@ -147,34 +147,26 @@ async def nested_server(): await asyncio.wait_for(server_task, timeout=2.0) -async def test_ping(streamable_http_server: str): +async def test_ping(streamable_http_server: ASGIServer): """Test pinging the server.""" - async with Client( - transport=StreamableHttpTransport(streamable_http_server) - ) as client: + async with streamable_http_server.client() as client: result = await client.ping() assert result is True async def test_ping_with_streamable_http_alias( - streamable_http_server_with_streamable_http_alias: str, + streamable_http_server_with_streamable_http_alias: ASGIServer, ): """Test pinging the server.""" - async with Client( - transport=StreamableHttpTransport( - streamable_http_server_with_streamable_http_alias - ) - ) as client: + async with streamable_http_server_with_streamable_http_alias.client() as client: result = await client.ping() assert result is True -async def test_http_headers(streamable_http_server: str): +async def test_http_headers(streamable_http_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=StreamableHttpTransport( - streamable_http_server, headers={"X-DEMO-HEADER": "ABC"} - ) + async with streamable_http_server.client( + headers={"X-DEMO-HEADER": "ABC"} ) as client: raw_result = await client.read_resource("request://headers") assert isinstance(raw_result[0], TextResourceContents) @@ -183,9 +175,9 @@ async def test_http_headers(streamable_http_server: str): assert json_result["x-demo-header"] == "ABC" -async def test_session_id_callback(streamable_http_server: str): +async def test_session_id_callback(streamable_http_server: ASGIServer): """Test getting mcp-session-id from the transport.""" - transport = StreamableHttpTransport(streamable_http_server) + transport = streamable_http_server.transport() assert transport.get_session_id() is None async with Client(transport=transport): session_id = transport.get_session_id() @@ -193,13 +185,12 @@ async def test_session_id_callback(streamable_http_server: str): @pytest.mark.parametrize("streamable_http_server", [True, False], indirect=True) -async def test_greet_with_progress_tool(streamable_http_server: str): +async def test_greet_with_progress_tool(streamable_http_server: ASGIServer): """Test calling the greet tool.""" progress_handler = AsyncMock(return_value=None) - async with Client( - transport=StreamableHttpTransport(streamable_http_server), - progress_handler=progress_handler, + async with streamable_http_server.client( + progress_handler=progress_handler ) as client: result = await client.call_tool("greet_with_progress", {"name": "Alice"}) assert result.data == "Hello, Alice!" @@ -213,7 +204,7 @@ async def test_greet_with_progress_tool(streamable_http_server: str): @pytest.mark.parametrize("streamable_http_server", [True, False], indirect=True) -async def test_elicitation_tool(streamable_http_server: str, request): +async def test_elicitation_tool(streamable_http_server: ASGIServer, request): """Test calling the elicitation tool in both stateless and stateful modes.""" async def elicitation_handler(message, response_type, params, ctx): @@ -223,30 +214,27 @@ async def test_elicitation_tool(streamable_http_server: str, request): if stateless_http: pytest.xfail("Elicitation is not supported in stateless HTTP mode") - async with Client( - transport=StreamableHttpTransport(streamable_http_server), - elicitation_handler=elicitation_handler, + async with streamable_http_server.client( + elicitation_handler=elicitation_handler ) as client: result = await client.call_tool("elicit") assert result.data == "You said your name was: Alice!" @pytest.mark.parametrize("streamable_http_server", [True], indirect=True) -async def test_stateless_http_rejects_get_sse(streamable_http_server: str): +async def test_stateless_http_rejects_get_sse(streamable_http_server: ASGIServer): """Stateless servers should reject GET SSE requests with 405.""" - import httpx2 - - async with httpx2.AsyncClient() as http_client: - response = await http_client.get(streamable_http_server) + async with streamable_http_server.http_client() as http_client: + response = await http_client.get(streamable_http_server.url) assert response.status_code == 405 @pytest.mark.parametrize("streamable_http_server", [True], indirect=True) -async def test_stateless_http_still_accepts_post(streamable_http_server: str): +async def test_stateless_http_still_accepts_post( + streamable_http_server: ASGIServer, +): """Stateless servers should still handle POST requests normally.""" - async with Client( - transport=StreamableHttpTransport(streamable_http_server) - ) as client: + async with streamable_http_server.client() as client: result = await client.call_tool("greet", {"name": "World"}) assert result.data == "Hello, World!" @@ -263,29 +251,21 @@ async def test_nested_streamable_http_server_resolves_correctly(nested_server: s reason="Timeout tests are flaky on Windows. Timeouts *are* supported but the tests are unreliable.", ) class TestTimeout: - async def test_timeout(self, streamable_http_server: str): + async def test_timeout(self, streamable_http_server: ASGIServer): # note this transport behaves differently than others and raises # MCPError from the *client* context with pytest.raises(MCPError, match="timed out"): - async with Client( - transport=StreamableHttpTransport(streamable_http_server), - timeout=0.02, - ) as client: + async with streamable_http_server.client(timeout=0.02) as client: await client.call_tool("sleep", {"seconds": 0.05}) - async def test_timeout_tool_call(self, streamable_http_server: str): - async with Client( - transport=StreamableHttpTransport(streamable_http_server), - ) as client: + async def test_timeout_tool_call(self, streamable_http_server: ASGIServer): + async with streamable_http_server.client() as client: with pytest.raises(MCPError): await client.call_tool("sleep", {"seconds": 0.2}, timeout=0.1) async def test_timeout_tool_call_overrides_client_timeout( - self, streamable_http_server: str + self, streamable_http_server: ASGIServer ): - async with Client( - transport=StreamableHttpTransport(streamable_http_server), - timeout=2, - ) as client: + async with streamable_http_server.client(timeout=2) as client: with pytest.raises(MCPError): await client.call_tool("sleep", {"seconds": 0.2}, timeout=0.1) diff --git a/tests/conformance/server.py b/tests/conformance/server.py index 158b0949e..9a6a97c28 100644 --- a/tests/conformance/server.py +++ b/tests/conformance/server.py @@ -116,9 +116,9 @@ async def test_error_handling() -> str: async def test_tool_with_logging(ctx: Context) -> str: """Sends log notifications during execution.""" await ctx.info("Tool execution started") - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) await ctx.info("Tool processing data") - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) await ctx.info("Tool execution completed") return "Logging test complete." @@ -127,9 +127,9 @@ async def test_tool_with_logging(ctx: Context) -> str: async def test_tool_with_progress(ctx: Context) -> str: """Reports progress notifications.""" await ctx.report_progress(0, 100) - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) await ctx.report_progress(50, 100) - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) await ctx.report_progress(100, 100) return "Progress test complete." diff --git a/tests/conformance/test_conformance.py b/tests/conformance/test_conformance.py index 971efbfa4..78749f8bc 100644 --- a/tests/conformance/test_conformance.py +++ b/tests/conformance/test_conformance.py @@ -55,7 +55,7 @@ def conformance_server(_require_npx): with socket.create_connection((HOST, port), timeout=1): break except OSError: - time.sleep(0.1) + time.sleep(0.01) else: pytest.fail("Conformance server did not start in time") diff --git a/tests/conftest.py b/tests/conftest.py index fcca01131..98f1c6aa9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -85,19 +85,42 @@ def enable_fastmcp_logger_propagation(caplog): root_logger.propagate = original_propagate +@pytest.fixture(scope="session") +def _settings_home_root(tmp_path_factory: pytest.TempPathFactory) -> Path: + """Session-scoped (i.e. per xdist-worker) base directory for isolated + settings.home directories. + + Created once via ``tmp_path_factory`` so ``isolate_settings_home`` can + carve out a per-test subdirectory with a plain, cheap ``mkdir`` instead + of requesting a fresh ``tmp_path`` (which every test would otherwise pay + for, autouse) on every single test. + """ + return tmp_path_factory.mktemp("fastmcp-test-home") + + @pytest.fixture(autouse=True) -def isolate_settings_home(tmp_path: Path): +def isolate_settings_home(_settings_home_root: Path): """Ensure each test uses an isolated settings.home directory. This prevents file locking issues when multiple tests share the same - storage directory in settings.home / "oauth-proxy". + storage directory in settings.home / "oauth-proxy". That collision is + not hypothetical: most oauth-proxy tests construct their proxy with the + same hardcoded jwt_signing_key ("test-secret"), and the storage + directory's name is a fingerprint derived from that key -- so any two + tests reusing it resolve to the *same* subdirectory. Reusing a single + settings.home across the whole session/worker would let one test's + persisted client/token state leak into the next, even though the tests + run sequentially within a worker. A fresh subdirectory per test avoids + that leakage while a session-scoped root avoids paying tmp_path's + per-test overhead (numbering, test-id sanitization, retention-policy + bookkeeping) for the ~99% of tests that never touch this directory. Also sets a fast Docket polling interval for tests — the default 50ms is fine for production but still adds ~25ms average pickup latency per task. 10ms makes task tests near-instant. """ - test_home = tmp_path / "fastmcp-test-home" - test_home.mkdir(exist_ok=True) + test_home = _settings_home_root / secrets.token_hex(8) + test_home.mkdir() with temporary_settings( home=test_home, diff --git a/tests/resources/test_resource_security.py b/tests/resources/test_resource_security.py index f530d63c9..ef56c6529 100644 --- a/tests/resources/test_resource_security.py +++ b/tests/resources/test_resource_security.py @@ -149,6 +149,7 @@ class TestBareSlimImport: The import must be deferred to the point of actual screening. """ + @pytest.mark.subprocess_heavy def test_resources_import_without_sdk(self): code = textwrap.dedent( """ diff --git a/tests/server/auth/oauth_proxy/conftest.py b/tests/server/auth/oauth_proxy/conftest.py index 4e402ad05..f3a02edbc 100644 --- a/tests/server/auth/oauth_proxy/conftest.py +++ b/tests/server/auth/oauth_proxy/conftest.py @@ -3,6 +3,7 @@ import asyncio import secrets import time +from contextlib import suppress from unittest.mock import Mock from urllib.parse import urlencode @@ -31,6 +32,7 @@ class MockOAuthProvider: self.base_url = f"http://localhost:{port}" self.app = None self.server = None + self._serve_task: asyncio.Task | None = None # Storage for OAuth state self.authorization_codes = {} @@ -235,16 +237,25 @@ class MockOAuthProvider: self.server = Server(config) # Start server in background - asyncio.create_task(self.server.serve()) + self._serve_task = asyncio.create_task(self.server.serve()) - # Wait for server to be ready - await asyncio.sleep(0.05) + # Wait for the server to finish startup instead of a fixed sleep. + # uvicorn.Server flips `started` to True once the listening socket + # is bound, right before it would start accepting connections. + deadline = asyncio.get_event_loop().time() + 5.0 + while not self.server.started: + if asyncio.get_event_loop().time() > deadline: + raise RuntimeError("Mock OAuth server failed to start in time") + await asyncio.sleep(0.005) async def stop(self): """Stop the mock OAuth server.""" if self.server: self.server.should_exit = True - await asyncio.sleep(0.01) + if self._serve_task is not None: + # Wait for the actual shutdown rather than a fixed sleep. + with suppress(TimeoutError): + await asyncio.wait_for(self._serve_task, timeout=5.0) def reset(self): """Reset all state for next test.""" diff --git a/tests/server/auth/oauth_proxy/test_identity_assertion.py b/tests/server/auth/oauth_proxy/test_identity_assertion.py index bf0495090..bbf463d34 100644 --- a/tests/server/auth/oauth_proxy/test_identity_assertion.py +++ b/tests/server/auth/oauth_proxy/test_identity_assertion.py @@ -207,6 +207,7 @@ class TestIdentityAssertionConfig: with pytest.raises(ValueError): IdentityAssertion(trusted_issuers=[ISSUER], algorithms={ISSUER: "HS256"}) + @pytest.mark.subprocess_heavy def test_lazy_reexport_does_not_import_module(self): # fastmcp.server.auth must not load identity_assertion (and its # httpx2 dependency) eagerly — the re-export is lazy via __getattr__. diff --git a/tests/server/auth/test_jwt_issuer.py b/tests/server/auth/test_jwt_issuer.py index 1f1b8d3ea..f34d96b9c 100644 --- a/tests/server/auth/test_jwt_issuer.py +++ b/tests/server/auth/test_jwt_issuer.py @@ -1,7 +1,6 @@ """Unit tests for JWT issuer and token encryption.""" import base64 -import time import pytest from joserfc.errors import JoseError @@ -163,24 +162,26 @@ class TestJWTIssuer: def test_verify_token_validates_expiration(self, issuer): """Test that expired tokens are rejected.""" - # Create token that expires in 1 second - token = issuer.issue_access_token( + # A token that is still valid should verify successfully. + valid_token = issuer.issue_access_token( client_id="client-abc", scopes=["read"], - jti="token-id", - expires_in=1, + jti="valid-token-id", ) - - # Should be valid immediately - payload = issuer.verify_token(token) + payload = issuer.verify_token(valid_token) assert payload["client_id"] == "client-abc" - # Wait for token to expire - time.sleep(1.1) - - # Should be rejected + # A token issued already-expired should be rejected. verify_token() + # does a strict `exp < time.time()` comparison with no clock-skew + # leeway, so this is instant and deterministic (no sleep needed). + expired_token = issuer.issue_access_token( + client_id="client-abc", + scopes=["read"], + jti="expired-token-id", + expires_in=-10, + ) with pytest.raises(JoseError, match="expired"): - issuer.verify_token(token) + issuer.verify_token(expired_token) def test_verify_token_validates_issuer(self, issuer): """Test that tokens from different issuers are rejected.""" diff --git a/tests/server/http/test_http_dependencies.py b/tests/server/http/test_http_dependencies.py index a2cd49774..d03b3df90 100644 --- a/tests/server/http/test_http_dependencies.py +++ b/tests/server/http/test_http_dependencies.py @@ -4,11 +4,9 @@ import pytest from mcp_types import TextContent, TextResourceContents from starlette.requests import Request -from fastmcp.client import Client -from fastmcp.client.transports import SSETransport, StreamableHttpTransport from fastmcp.server.dependencies import CurrentHeaders, CurrentRequest, get_http_request from fastmcp.server.server import FastMCP -from fastmcp.utilities.tests import run_server_async +from fastmcp.utilities.tests import ASGIServer, asgi_server def fastmcp_server(): @@ -44,25 +42,21 @@ def fastmcp_server(): async def shttp_server(): """Start a test server with StreamableHttp transport.""" server = fastmcp_server() - async with run_server_async(server, transport="http") as url: - yield url + async with asgi_server(server, transport="http") as running_server: + yield running_server @pytest.fixture async def sse_server(): """Start a test server with SSE transport.""" server = fastmcp_server() - async with run_server_async(server, transport="sse") as url: - yield url + async with asgi_server(server, transport="sse") as running_server: + yield running_server -async def test_http_headers_resource_shttp(shttp_server: str): +async def test_http_headers_resource_shttp(shttp_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=StreamableHttpTransport( - shttp_server, headers={"X-DEMO-HEADER": "ABC"} - ) - ) as client: + async with shttp_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: raw_result = await client.read_resource("request://headers") assert isinstance(raw_result[0], TextResourceContents) json_result = json.loads(raw_result[0].text) @@ -70,11 +64,9 @@ async def test_http_headers_resource_shttp(shttp_server: str): assert json_result["x-demo-header"] == "ABC" -async def test_http_headers_resource_sse(sse_server: str): +async def test_http_headers_resource_sse(sse_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"}) - ) as client: + async with sse_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: raw_result = await client.read_resource("request://headers") assert isinstance(raw_result[0], TextResourceContents) json_result = json.loads(raw_result[0].text) @@ -82,34 +74,24 @@ async def test_http_headers_resource_sse(sse_server: str): assert json_result["x-demo-header"] == "ABC" -async def test_http_headers_tool_shttp(shttp_server: str): +async def test_http_headers_tool_shttp(shttp_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=StreamableHttpTransport( - shttp_server, headers={"X-DEMO-HEADER": "ABC"} - ) - ) as client: + async with shttp_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: result = await client.call_tool("get_headers_tool") assert "x-demo-header" in result.data assert result.data["x-demo-header"] == "ABC" -async def test_http_headers_tool_sse(sse_server: str): - async with Client( - transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"}) - ) as client: +async def test_http_headers_tool_sse(sse_server: ASGIServer): + async with sse_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: result = await client.call_tool("get_headers_tool") assert "x-demo-header" in result.data assert result.data["x-demo-header"] == "ABC" -async def test_http_headers_prompt_shttp(shttp_server: str): +async def test_http_headers_prompt_shttp(shttp_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=StreamableHttpTransport( - shttp_server, headers={"X-DEMO-HEADER": "ABC"} - ) - ) as client: + async with shttp_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: result = await client.get_prompt("get_headers_prompt") assert isinstance(result.messages[0].content, TextContent) json_result = json.loads(result.messages[0].content.text) @@ -117,11 +99,9 @@ async def test_http_headers_prompt_shttp(shttp_server: str): assert json_result["x-demo-header"] == "ABC" -async def test_http_headers_prompt_sse(sse_server: str): +async def test_http_headers_prompt_sse(sse_server: ASGIServer): """Test getting HTTP headers from the server.""" - async with Client( - transport=SSETransport(sse_server, headers={"X-DEMO-HEADER": "ABC"}) - ) as client: + async with sse_server.client(headers={"X-DEMO-HEADER": "ABC"}) as client: result = await client.get_prompt("get_headers_prompt") assert isinstance(result.messages[0].content, TextContent) json_result = json.loads(result.messages[0].content.text) @@ -129,7 +109,7 @@ async def test_http_headers_prompt_sse(sse_server: str): assert json_result["x-demo-header"] == "ABC" -async def test_get_http_headers_excludes_content_type(sse_server: str): +async def test_get_http_headers_excludes_content_type(sse_server: ASGIServer): """Test that get_http_headers() excludes content-type header (issue #3097). This prevents HTTP 415 errors when forwarding headers to downstream APIs @@ -144,16 +124,13 @@ async def test_get_http_headers_excludes_content_type(sse_server: str): """Check that problematic headers are excluded from get_http_headers().""" return get_http_headers() - async with run_server_async(server, transport="sse") as url: - async with Client( - transport=SSETransport( - url, - headers={ - "Content-Type": "application/json", - "Accept": "application/json", - "X-Custom-Header": "should-be-included", - }, - ) + async with asgi_server(server, transport="sse") as running_server: + async with running_server.client( + headers={ + "Content-Type": "application/json", + "Accept": "application/json", + "X-Custom-Header": "should-be-included", + } ) as client: result = await client.call_tool("check_excluded_headers") headers = result.data @@ -178,9 +155,9 @@ async def test_background_task_can_read_snapshotted_request_headers(): request = get_http_request() return request.headers.get("x-tenant-id", "missing") - async with run_server_async(server, transport="sse") as url: - async with Client( - transport=SSETransport(url, headers={"X-Tenant-ID": "tenant-123"}) + async with asgi_server(server, transport="sse") as running_server: + async with running_server.client( + headers={"X-Tenant-ID": "tenant-123"} ) as client: task = await client.call_tool("check_request_header", task=True) result = await task.result() @@ -201,15 +178,12 @@ async def test_background_task_current_http_dependencies_restore_headers(): "tenant": request.headers.get("x-tenant-id", "missing"), } - async with run_server_async(server, transport="sse") as url: - async with Client( - transport=SSETransport( - url, - headers={ - "Authorization": "Bearer tenant-token", - "X-Tenant-ID": "tenant-456", - }, - ) + async with asgi_server(server, transport="sse") as running_server: + async with running_server.client( + headers={ + "Authorization": "Bearer tenant-token", + "X-Tenant-ID": "tenant-456", + } ) as client: task = await client.call_tool("check_headers", task=True) result = await task.result() diff --git a/tests/server/http/test_session_idle_timeout.py b/tests/server/http/test_session_idle_timeout.py index 6b3558723..f7adad149 100644 --- a/tests/server/http/test_session_idle_timeout.py +++ b/tests/server/http/test_session_idle_timeout.py @@ -43,7 +43,7 @@ def test_idle_session_is_terminated_after_timeout(): app = create_streamable_http_app( server=server, streamable_http_path="/mcp", - session_idle_timeout=0.2, + session_idle_timeout=0.1, ) with TestClient(app, base_url="http://127.0.0.1") as client: @@ -58,11 +58,15 @@ def test_idle_session_is_terminated_after_timeout(): # Wait past the idle deadline; the SDK's idle cancel scope fires and # removes the session from the active instances. Poll to stay fast. + # The idle timeout itself is driven by anyio's event-loop clock + # inside the SDK (not a mockable Python-level time source), so this + # remains a real wait; the timeout and poll interval are kept as + # small as reliably possible. deadline = time.monotonic() + 3.0 while time.monotonic() < deadline: if session_id not in sm._server_instances: break - time.sleep(0.05) + time.sleep(0.02) assert session_id not in sm._server_instances diff --git a/tests/server/middleware/test_initialization_middleware.py b/tests/server/middleware/test_initialization_middleware.py index 3a077b367..d65a72c24 100644 --- a/tests/server/middleware/test_initialization_middleware.py +++ b/tests/server/middleware/test_initialization_middleware.py @@ -381,9 +381,8 @@ async def test_state_isolation_between_streamable_http_clients(): Each client should have its own session ID and isolated state. """ - from fastmcp.client.transports import StreamableHttpTransport from fastmcp.server.context import Context - from fastmcp.utilities.tests import run_server_async + from fastmcp.utilities.tests import asgi_server server = FastMCP("TestServer") @@ -398,12 +397,11 @@ async def test_state_isolation_between_streamable_http_clients(): "session_id": ctx.session_id, } - async with run_server_async(server, transport="streamable-http") as url: + async with asgi_server(server, transport="streamable-http") as running_server: import json # Client 1 stores its value - transport1 = StreamableHttpTransport(url=url) - async with Client(transport=transport1) as client1: + async with running_server.client() as client1: result1 = await client1.call_tool( "store_and_read", {"value": "client1-value"} ) @@ -413,8 +411,7 @@ async def test_state_isolation_between_streamable_http_clients(): session_id_1 = data1["session_id"] # Client 2 should have completely isolated state - transport2 = StreamableHttpTransport(url=url) - async with Client(transport=transport2) as client2: + async with running_server.client() as client2: result2 = await client2.call_tool( "store_and_read", {"value": "client2-value"} ) diff --git a/tests/server/middleware/test_rate_limiting.py b/tests/server/middleware/test_rate_limiting.py index 4ae5ae114..525107ea4 100644 --- a/tests/server/middleware/test_rate_limiting.py +++ b/tests/server/middleware/test_rate_limiting.py @@ -60,8 +60,14 @@ class TestTokenBucketRateLimiter: # Should fail to consume more assert await limiter.consume(1) is False - async def test_refill(self): + async def test_refill(self, monkeypatch): """Test token refill over time.""" + current_time = 0.0 + monkeypatch.setattr( + "fastmcp.server.middleware.rate_limiting.time.time", + lambda: current_time, + ) + limiter = TokenBucketRateLimiter( capacity=10, refill_rate=10.0 ) # 10 tokens per second @@ -70,11 +76,13 @@ class TestTokenBucketRateLimiter: assert await limiter.consume(10) is True assert await limiter.consume(1) is False - # Wait for refill (0.2 seconds = 2 tokens at 10/sec) - await asyncio.sleep(0.2) + # Advance the clock instead of sleeping in real time. Use a small + # base time and a slight margin over the strict 0.2s/2-token + # threshold so the result isn't sensitive to float rounding. + current_time += 0.25 assert await limiter.consume(2) is True - async def test_denied_consumes_do_not_freeze_clock(self): + async def test_denied_consumes_do_not_freeze_clock(self, monkeypatch): """Regression for #4056: a client that retries quickly after being denied must not be able to bypass the configured refill rate. @@ -82,21 +90,26 @@ class TestTokenBucketRateLimiter: If it only advanced on success, the elapsed window would be re-counted on each retry, letting a client refill faster than `refill_rate`. """ + current_time = 0.0 + monkeypatch.setattr( + "fastmcp.server.middleware.rate_limiting.time.time", + lambda: current_time, + ) + limiter = TokenBucketRateLimiter(capacity=10, refill_rate=10.0) # Drain the bucket. assert await limiter.consume(10) is True - # Hammer with denied requests over ~0.2s. With the correct - # implementation, last_refill advances on each call, so total - # accumulated tokens after 0.2s is ~2 (10/s * 0.2s). + # Hammer with denied requests over a simulated ~0.2s (no real sleep). + # With the correct implementation, last_refill advances on each + # call, so total accumulated tokens after 0.2s is ~2 (10/s * 0.2s). for _ in range(20): await limiter.consume(1) - await asyncio.sleep(0.01) + current_time += 0.01 # We should NOT be able to consume more than the configured rate - # would allow over the elapsed window. Allow a small slack for - # timing jitter, but stay well below `capacity`. + # would allow over the elapsed window, well below `capacity`. assert await limiter.consume(5) is False, ( "denied retries should not silently accrue extra tokens" ) @@ -132,8 +145,14 @@ class TestSlidingWindowRateLimiter: # Should reject over limit assert await limiter.is_allowed() is False - async def test_sliding_window(self): + async def test_sliding_window(self, monkeypatch): """Test sliding window behavior.""" + current_time = 0.0 + monkeypatch.setattr( + "fastmcp.server.middleware.rate_limiting.time.time", + lambda: current_time, + ) + limiter = SlidingWindowRateLimiter(max_requests=2, window_seconds=1) # Use up requests @@ -141,8 +160,8 @@ class TestSlidingWindowRateLimiter: assert await limiter.is_allowed() is True assert await limiter.is_allowed() is False - # Wait for window to pass - await asyncio.sleep(1.1) + # Advance the clock past the window instead of sleeping in real time + current_time += 1.1 # Should be able to make requests again assert await limiter.is_allowed() is True @@ -528,12 +547,11 @@ class TestRateLimitingMiddlewareIntegration: async def test_rate_limiting_recovery_over_time(self, rate_limit_server): """Test that rate limiting allows requests again after time passes.""" - rate_limit_server.add_middleware( - RateLimitingMiddleware( - max_requests_per_second=10.0, # 10 per second = 1 every 100ms - burst_capacity=4, - ) + middleware = RateLimitingMiddleware( + max_requests_per_second=10.0, # 10 per second = 1 every 100ms + burst_capacity=4, ) + rate_limit_server.add_middleware(middleware) async with Client(rate_limit_server) as client: # Exhaust the burst; the exact number of internal requests before the @@ -547,8 +565,11 @@ class TestRateLimitingMiddlewareIntegration: break assert hit_limit, "Rate limit was never triggered" - # Wait for token bucket to refill (150ms should be enough for ~1.5 tokens) - await asyncio.sleep(0.15) + # Simulate token refill without a real sleep: rewind each + # bucket's last-refill timestamp so the next consume() sees + # ~150ms of elapsed time (10 tokens/sec => ~1.5 tokens refilled). + for limiter in middleware.limiters.values(): + limiter.last_refill -= 0.15 # Should be able to make another request result = await client.call_tool("quick_action", {"message": "after_wait"}) diff --git a/tests/server/tasks/test_concurrent_dependencies.py b/tests/server/tasks/test_concurrent_dependencies.py index eb5af1bb1..7f6f79563 100644 --- a/tests/server/tasks/test_concurrent_dependencies.py +++ b/tests/server/tasks/test_concurrent_dependencies.py @@ -26,7 +26,7 @@ async def test_concurrent_foreground_tools_with_context(): @mcp.tool() async def slow_tool(name: str, ctx: Context) -> str: - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) results.append(name) return f"done:{name}" @@ -77,7 +77,7 @@ async def test_concurrent_background_tasks_with_context(): @mcp.tool(task=True) async def bg_tool(name: str, ctx: Context) -> str: - await asyncio.sleep(0.05) + await asyncio.sleep(0.01) return f"bg:{name}" async with Client(mcp) as client: diff --git a/tests/server/tasks/test_notifications.py b/tests/server/tasks/test_notifications.py index ca7abfe1a..a9dfcf00d 100644 --- a/tests/server/tasks/test_notifications.py +++ b/tests/server/tasks/test_notifications.py @@ -6,6 +6,7 @@ No mocking of Redis, sessions, or Docket internals. """ import asyncio +import time import mcp_types @@ -123,8 +124,7 @@ class TestNotificationIntegration: # After client disconnects, subscriber should be cleaned up # Allow brief time for async cleanup - for _ in range(20): - if get_subscriber_count() == count_before: - break - await asyncio.sleep(0.05) + deadline = time.monotonic() + 1.0 + while get_subscriber_count() != count_before and time.monotonic() < deadline: + await asyncio.sleep(0.005) assert get_subscriber_count() == count_before diff --git a/tests/server/tasks/test_progress_dependency.py b/tests/server/tasks/test_progress_dependency.py index 6cc35996a..2c56f2f22 100644 --- a/tests/server/tasks/test_progress_dependency.py +++ b/tests/server/tasks/test_progress_dependency.py @@ -77,9 +77,10 @@ async def test_progress_status_message_in_background_task(): await progress.increment() step_started.set() - # Give test time to poll status - await asyncio.sleep(0.2) - + # No settling wait needed: the server never clears the progress + # message on completion (only a failure overwrites it), so whatever + # "Step N of 3" message is current when the test polls status() + # below still satisfies the assertion, win or lose the race. await progress.set_message("Step 2 of 3") await progress.increment() await progress.set_message("Step 3 of 3") diff --git a/tests/server/tasks/test_task_methods.py b/tests/server/tasks/test_task_methods.py index 493bc9167..e00f1ced6 100644 --- a/tests/server/tasks/test_task_methods.py +++ b/tests/server/tasks/test_task_methods.py @@ -5,6 +5,7 @@ Tests the tasks/get, tasks/result, and tasks/list JSON-RPC protocol methods. """ import asyncio +import time import pytest from mcp.shared.exceptions import MCPError @@ -30,8 +31,12 @@ async def endpoint_server(): @mcp.tool(task=True) # Enable background execution async def slow_tool() -> str: - """A slow tool for testing cancellation.""" - await asyncio.sleep(10) + """A slow tool for testing cancellation. + + Never completes on its own - the only test that submits this task + cancels it well before any real-time completion would matter. + """ + await asyncio.Event().wait() return "done" return mcp @@ -160,17 +165,24 @@ async def test_task_cancellation_workflow(endpoint_server): # Submit slow task task = await client.call_tool("slow_tool", {}, task=True) - # Give it a moment to start - await asyncio.sleep(0.1) + # Wait until the task is tracked as working before cancelling + deadline = time.monotonic() + 5.0 + status = await task.status() + while status.status != "working" and time.monotonic() < deadline: + await asyncio.sleep(0.005) + status = await task.status() # Cancel the task await task.cancel() - # Give cancellation a moment to process - await asyncio.sleep(0.1) + # Poll until cancellation is reflected in task status + deadline = time.monotonic() + 5.0 + status = await task.status() + while status.status != "cancelled" and time.monotonic() < deadline: + await asyncio.sleep(0.005) + status = await task.status() # Task should be in cancelled state - status = await task.status() assert status.status == "cancelled" @@ -192,7 +204,11 @@ async def test_task_cancellation_interrupts_running_coroutine(endpoint_server): async def interruptible_tool() -> str: started.set() try: - await asyncio.sleep(60) + # Never completes on its own - the test cancels this task well + # before any real-time completion would matter, so a genuinely + # suspended coroutine (rather than a fixed-duration sleep) is + # enough to prove cancellation delivers CancelledError. + await asyncio.Event().wait() completed_normally.set() return "completed" except asyncio.CancelledError: diff --git a/tests/server/tasks/test_task_mount.py b/tests/server/tasks/test_task_mount.py index 0401259c1..bc82c7188 100644 --- a/tests/server/tasks/test_task_mount.py +++ b/tests/server/tasks/test_task_mount.py @@ -6,6 +6,7 @@ on mounted child servers through a parent server. """ import asyncio +import time import mcp_types as mt import pytest @@ -140,7 +141,7 @@ class TestMountedToolTasks: """Can poll task status for mounted tool.""" async with Client(parent_server) as client: task = await client.call_tool( - "child_slow_child_tool", {"duration": 0.5}, task=True + "child_slow_child_tool", {"duration": 0.05}, task=True ) # Check status while running @@ -162,15 +163,28 @@ class TestMountedToolTasks: "child_slow_child_tool", {"duration": 10.0}, task=True ) - # Let it start - await asyncio.sleep(0.1) + # Wait until the task is tracked as working before cancelling it + deadline = time.monotonic() + 5.0 + status = await task.status() + while status.status != "working" and time.monotonic() < deadline: + await asyncio.sleep(0.005) + status = await task.status() # Cancel the task await task.cancel() - # Check status + # Cancellation propagation isn't instantaneous, so poll for the + # terminal state rather than asserting immediately after cancel(). + deadline = time.monotonic() + 5.0 status = await task.status() - assert status.status == "cancelled" + while status.status != "cancelled" and time.monotonic() < deadline: + await asyncio.sleep(0.005) + status = await task.status() + + assert status.status == "cancelled", ( + f"task did not reach 'cancelled' within 5s of cancel() " + f"(last status: {status.status!r})" + ) async def test_graceful_degradation_sync_mounted_tool(self, parent_server): """Sync-only mounted tool returns error with task=True.""" diff --git a/tests/server/tasks/test_task_status_notifications.py b/tests/server/tasks/test_task_status_notifications.py index 98d333ca9..5fb4cec7d 100644 --- a/tests/server/tasks/test_task_status_notifications.py +++ b/tests/server/tasks/test_task_status_notifications.py @@ -28,9 +28,14 @@ async def notification_server(): return value * 2 @mcp.tool(task=True) - async def slow_task(duration: float = 0.1) -> str: - """Slow task for testing working status.""" - await asyncio.sleep(duration) + async def slow_task() -> str: + """Task that never completes on its own. + + Only used to verify disconnect-while-running doesn't crash - the + test disconnects before the task would finish, so it never needs + to actually complete. + """ + await asyncio.Event().wait() return "completed" @mcp.tool(task=True) @@ -41,13 +46,11 @@ async def notification_server(): @mcp.prompt(task=True) async def test_prompt(name: str) -> str: """Test prompt for background execution.""" - await asyncio.sleep(0.05) return f"Hello, {name}!" @mcp.resource("test://resource", task=True) async def test_resource() -> str: """Test resource for background execution.""" - await asyncio.sleep(0.05) return "resource content" return mcp @@ -84,9 +87,9 @@ async def test_subscription_handles_task_completion(notification_server: FastMCP assert result2.data == 4 assert result3.data == 6 - # Subscriptions should all clean up - # Give them a moment - await asyncio.sleep(0.1) + # Subscriptions clean up deterministically via the connection's + # exit stack when the client disconnects (see test below), so no + # settling wait is needed here. async def test_subscription_handles_task_failure(notification_server: FastMCP): @@ -98,8 +101,8 @@ async def test_subscription_handles_task_failure(notification_server: FastMCP): with pytest.raises(Exception): await task - # Subscription should handle failure and clean up - await asyncio.sleep(0.1) + # Subscription cleans up deterministically via the connection's + # exit stack on disconnect; no settling wait is needed here. async def test_subscription_for_prompt_tasks(notification_server: FastMCP): @@ -111,8 +114,8 @@ async def test_subscription_for_prompt_tasks(notification_server: FastMCP): # Prompt result has messages assert result - # Subscription should clean up - await asyncio.sleep(0.1) + # Subscription cleans up deterministically via the connection's + # exit stack on disconnect; no settling wait is needed here. async def test_subscription_for_resource_tasks(notification_server: FastMCP): @@ -123,8 +126,8 @@ async def test_subscription_for_resource_tasks(notification_server: FastMCP): result = await task assert result # Resource contents - # Subscription should clean up - await asyncio.sleep(0.1) + # Subscription cleans up deterministically via the connection's + # exit stack on disconnect; no settling wait is needed here. async def test_subscriptions_cleanup_on_session_disconnect( @@ -133,7 +136,7 @@ async def test_subscriptions_cleanup_on_session_disconnect( """Subscriptions are cleaned up when session disconnects.""" # Start session and create task async with Client(notification_server) as client: - task = await client.call_tool("slow_task", {"duration": 1.0}, task=True) + task = await client.call_tool("slow_task", {}, task=True) task_id = task.task_id # Disconnect before task completes (session __aexit__ cancels subscriptions) @@ -156,5 +159,5 @@ async def test_multiple_concurrent_subscriptions(notification_server: FastMCP): results = await asyncio.gather(*tasks) assert len(results) == 10 - # All subscriptions should clean up - await asyncio.sleep(0.1) + # All subscriptions clean up deterministically via the connection's + # exit stack on disconnect; no settling wait is needed here. diff --git a/tests/server/tasks/test_task_ttl.py b/tests/server/tasks/test_task_ttl.py index 769fafffa..1e297a7c8 100644 --- a/tests/server/tasks/test_task_ttl.py +++ b/tests/server/tasks/test_task_ttl.py @@ -24,7 +24,10 @@ async def keepalive_server(): @mcp.tool(task=True) async def slow_task() -> str: - await asyncio.sleep(1) + # Never completes during the test - the only test that submits this + # task checks status immediately after submission and never awaits + # completion, so there's no need for a real-time sleep here. + await asyncio.Event().wait() return "done" return mcp diff --git a/tests/server/test_context.py b/tests/server/test_context.py index 142287b3a..384a307ac 100644 --- a/tests/server/test_context.py +++ b/tests/server/test_context.py @@ -498,9 +498,7 @@ class TestTransportIntegration: async def test_transport_set_via_http_middleware(self): """Test that transport is set per-request via HTTP middleware.""" - from fastmcp import Client - from fastmcp.client.transports import StreamableHttpTransport - from fastmcp.utilities.tests import run_server_async + from fastmcp.utilities.tests import asgi_client mcp = FastMCP("test") observed_transport = None @@ -511,9 +509,7 @@ class TestTransportIntegration: observed_transport = ctx.transport return observed_transport or "none" - async with run_server_async(mcp, transport="streamable-http") as url: - transport = StreamableHttpTransport(url=url) - async with Client(transport=transport) as client: - result = await client.call_tool("get_transport", {}) - assert observed_transport == "streamable-http" - assert result.data == "streamable-http" + async with asgi_client(mcp, transport="streamable-http") as client: + result = await client.call_tool("get_transport", {}) + assert observed_transport == "streamable-http" + assert result.data == "streamable-http" diff --git a/tests/test_mcp_config.py b/tests/test_mcp_config.py index 9c028da0f..716e31aba 100644 --- a/tests/test_mcp_config.py +++ b/tests/test_mcp_config.py @@ -41,17 +41,24 @@ from fastmcp.mcp_config import ( from fastmcp.server.elicitation import AcceptedElicitation from fastmcp.tools.base import Tool as FastMCPTool -# These tests spawn subprocess servers via stdio which can be slow under -# parallel CI load. Give them more headroom than the 5s default, and skip -# entirely on Windows due to process lifecycle issues. +# Some tests in this module spawn subprocess servers via stdio, each paying a +# full interpreter startup plus `import fastmcp` (~0.7s). They take 3-6s idle, +# but on a loaded CI runner with four xdist workers competing they have blown a +# 15s ceiling. The timeout is here to catch a genuine hang, not to police speed, +# so give the module room rather than tuning each test individually. pytestmark = [ - pytest.mark.timeout(15), - pytest.mark.skipif( - sys.platform.startswith("win32"), - reason="Windows has process lifecycle issues with stdio subprocesses", - ), + pytest.mark.timeout(60), ] +# Most tests below run entirely in-memory (via InMemoryStdioMCPServer) or +# only parse/serialize config objects, so they're safe on Windows. Apply this +# marker only to tests that spawn a real subprocess (or attempt to, e.g. via +# a nonexistent command) — those still hit Windows process lifecycle issues. +requires_subprocess = pytest.mark.skipif( + sys.platform.startswith("win32"), + reason="Windows has process lifecycle issues with stdio subprocesses", +) + def running_under_debugger(): return os.environ.get("DEBUGPY_RUNNING") == "true" @@ -401,12 +408,13 @@ async def _wait_for_process_exit(pid: int, timeout: float = 3.0) -> None: psutil.Process(pid) except psutil.NoSuchProcess: return - await asyncio.sleep(0.05) + await asyncio.sleep(0.005) # Final check — if still alive, let the NoSuchProcess propagation fail the test clearly psutil.Process(pid) pytest.fail(f"Process {pid} still alive after {timeout}s") +@requires_subprocess @pytest.mark.skipif( running_under_debugger(), reason="Debugger holds a reference to the transport", @@ -467,6 +475,7 @@ async def test_multi_client_lifespan(tmp_path: Path): await _wait_for_process_exit(pid_2) +@requires_subprocess @pytest.mark.timeout(15) async def test_multi_client_force_close(tmp_path: Path): server_script = inspect.cleandoc(""" @@ -625,6 +634,7 @@ async def test_multi_client_with_logging(caplog): assert test_records[0].msg == "test 42" +@requires_subprocess async def test_multi_client_with_transforms(tmp_path: Path): """ Tests that transforms are properly applied to the tools. @@ -681,6 +691,7 @@ async def test_multi_client_with_transforms(tmp_path: Path): assert result.data == 3 +@requires_subprocess async def test_canonical_multi_client_with_transforms(tmp_path: Path): """Test that transforms are not applied to servers in a canonical MCPConfig.""" server_script = inspect.cleandoc(""" @@ -732,6 +743,7 @@ async def test_canonical_multi_client_with_transforms(tmp_path: Path): assert "test_1_transformed_add" not in tools_by_name +@requires_subprocess @pytest.mark.flaky(retries=3) async def test_multi_client_transform_with_filtering(tmp_path: Path): """ @@ -794,6 +806,7 @@ async def test_multi_client_transform_with_filtering(tmp_path: Path): assert "test_2_subtract" in tools_by_name +@requires_subprocess @pytest.mark.flaky(retries=3) async def test_single_server_config_include_tags_filtering(tmp_path: Path): """include_tags should filter tools even with a single server in the config.""" @@ -1012,6 +1025,7 @@ async def test_single_server_config_transport(): assert len(transport._transports) == 1 +@requires_subprocess @pytest.mark.parametrize( "server_order", [ @@ -1040,6 +1054,7 @@ async def test_multi_server_partial_failure(server_order: dict): assert len(tools) == 1 +@requires_subprocess async def test_multi_server_partial_failure_logs_warning(caplog): """A warning should be logged when a server fails to connect.""" config = MCPConfig( @@ -1064,6 +1079,7 @@ async def test_multi_server_partial_failure_logs_warning(caplog): assert len(warning_records) == 1 +@requires_subprocess async def test_multi_server_all_fail(): """When all servers fail to connect, a ConnectionError should be raised.""" config = MCPConfig( @@ -1095,6 +1111,7 @@ def _make_ping_server() -> FastMCP: return app +@requires_subprocess async def test_multi_server_partial_failure_cleanup(): """Transports for failed servers should not leak into _transports.""" config = MCPConfig( diff --git a/tests/test_no_legacy_httpx.py b/tests/test_no_legacy_httpx.py index e4cfb8c6e..5cd439f87 100644 --- a/tests/test_no_legacy_httpx.py +++ b/tests/test_no_legacy_httpx.py @@ -17,6 +17,8 @@ import subprocess import sys import textwrap +import pytest + _BLOCKER_SCRIPT = textwrap.dedent( """ import sys @@ -44,6 +46,7 @@ _BLOCKER_SCRIPT = textwrap.dedent( ) +@pytest.mark.subprocess_heavy def test_fastmcp_imports_without_legacy_httpx(): result = subprocess.run( [sys.executable, "-c", _BLOCKER_SCRIPT], diff --git a/tests/tools/test_standalone_decorator.py b/tests/tools/test_standalone_decorator.py index 4d57919cd..6b79001cc 100644 --- a/tests/tools/test_standalone_decorator.py +++ b/tests/tools/test_standalone_decorator.py @@ -29,6 +29,7 @@ from fastmcp.tools.function_tool import DecoratedTool, FunctionTool, ToolMeta "from fastmcp.server import Context, FastMCP, create_proxy", ], ) +@pytest.mark.subprocess_heavy def test_component_import_works_in_fresh_interpreter(statement: str): result = subprocess.run( [sys.executable, "-c", statement], diff --git a/tests/tools/test_tool_run_in_thread.py b/tests/tools/test_tool_run_in_thread.py index 069024905..932399cf4 100644 --- a/tests/tools/test_tool_run_in_thread.py +++ b/tests/tools/test_tool_run_in_thread.py @@ -142,7 +142,7 @@ class TestRunInThread: def blocking() -> str: import time - time.sleep(0.2) + time.sleep(0.05) return "done" ticks = 0 @@ -163,8 +163,10 @@ class TestRunInThread: assert isinstance(result.content[0], TextContent) assert result.content[0].text == "done" - # With inline execution, ticks should be near zero — the 200ms sleep - # blocks the loop. Under a thread pool (default), ticks would be ~10. + # With inline execution, ticks should be near zero — the blocking + # sleep never yields control, so at most one already-scheduled timer + # fires once control returns. Under a thread pool (default), ticks + # would scale with sleep duration instead. assert ticks <= 2 async def test_default_threadpool_permits_concurrency(self): @@ -196,7 +198,11 @@ class TestRunInThread: assert isinstance(result.content[0], TextContent) assert result.content[0].text == "done" - assert ticks >= 5 + # Ideal is 0.2s / 0.02s = 10 ticks. We only require 3 (30% of ideal) + # to tolerate a loaded/slow runner, while staying well clear of the + # blocked case's `ticks <= 2` bound above so the two tests can never + # produce overlapping, ambiguous results. + assert ticks >= 3 class TestRunInThreadViaStandaloneDecorator: diff --git a/tests/tools/test_tool_timeout.py b/tests/tools/test_tool_timeout.py index 301986f92..43532cbee 100644 --- a/tests/tools/test_tool_timeout.py +++ b/tests/tools/test_tool_timeout.py @@ -46,7 +46,7 @@ class TestToolTimeout: @mcp.tool(timeout=5.0) async def fast_async_tool() -> str: - await anyio.sleep(0.1) + await anyio.sleep(0.01) return "completed" result = await mcp.call_tool("fast_async_tool") @@ -59,7 +59,7 @@ class TestToolTimeout: @mcp.tool(timeout=5.0) def fast_sync_tool() -> str: - time.sleep(0.1) + time.sleep(0.01) return "completed" result = await mcp.call_tool("fast_sync_tool") @@ -72,7 +72,7 @@ class TestToolTimeout: @mcp.tool(timeout=0.2) async def slow_async_tool() -> str: - await anyio.sleep(2.0) + await anyio.sleep(0.6) return "should not reach" # TimeoutError is caught and converted to ToolError by FastMCP @@ -97,7 +97,7 @@ class TestToolTimeout: @mcp.tool(timeout=0.1) async def slow_tool() -> str: - await anyio.sleep(1.0) + await anyio.sleep(0.3) return "never" # Verify that ToolError is raised (timeout warning is logged to stderr) @@ -109,7 +109,7 @@ class TestToolTimeout: from fastmcp.tools import Tool async def my_slow_tool() -> str: - await anyio.sleep(1.0) + await anyio.sleep(0.3) return "never" tool = Tool.from_function(my_slow_tool, timeout=0.1) @@ -137,7 +137,7 @@ class TestToolTimeout: @mcp.tool(task=True, timeout=1.0) async def task_with_timeout() -> str: - await anyio.sleep(0.1) + await anyio.sleep(0.01) return "completed" # Tool should be registered successfully @@ -153,17 +153,17 @@ class TestToolTimeout: @mcp.tool(timeout=1.0) async def short_timeout() -> str: - await anyio.sleep(0.1) + await anyio.sleep(0.01) return "short" @mcp.tool(timeout=5.0) async def long_timeout() -> str: - await anyio.sleep(0.1) + await anyio.sleep(0.01) return "long" @mcp.tool async def no_timeout() -> str: - await anyio.sleep(0.1) + await anyio.sleep(0.01) return "none" # All should complete successfully @@ -184,7 +184,7 @@ class TestToolTimeout: @mcp.tool(timeout=0.1) async def times_out() -> str: - await anyio.sleep(1.0) + await anyio.sleep(0.3) return "never" # TimeoutError should be caught and converted to ToolError diff --git a/tests/utilities/test_asgi_transport.py b/tests/utilities/test_asgi_transport.py new file mode 100644 index 000000000..8260defb6 --- /dev/null +++ b/tests/utilities/test_asgi_transport.py @@ -0,0 +1,250 @@ +"""Tests for the in-process ASGI bridge and `asgi_server`.""" + +from typing import Literal + +import httpx2 +import pytest +from starlette.applications import Starlette +from starlette.requests import Request +from starlette.responses import Response, StreamingResponse +from starlette.routing import Route +from starlette.types import Receive, Scope, Send + +from fastmcp import Context, FastMCP +from fastmcp.server.auth.providers.jwt import JWTVerifier, RSAKeyPair +from fastmcp.utilities.asgi_transport import StreamingASGITransport, run_asgi_lifespan +from fastmcp.utilities.tests import ASGIServer, asgi_server + + +def build_server() -> FastMCP: + server = FastMCP("BridgeTestServer") + + @server.tool + def greet(name: str) -> str: + return f"Hello, {name}!" + + @server.tool + async def elicit_name(ctx: Context) -> str: + """Round-trips a server-initiated request while the response is still open.""" + result = await ctx.elicit("What is your name?", response_type=str) + if result.action == "accept": + return f"You said {result.data}" + return "declined" + + return server + + +class TestStreamingASGITransport: + async def test_forwards_chunks_as_the_app_produces_them(self): + """The bridge must stream, not buffer: chunks arrive before the app finishes.""" + + async def stream(request: Request) -> StreamingResponse: + async def body(): + yield b"first" + yield b"second" + + return StreamingResponse(body(), media_type="text/plain") + + app = Starlette(routes=[Route("/stream", stream)]) + async with httpx2.AsyncClient( + transport=StreamingASGITransport(app), base_url="http://testserver" + ) as client: + chunks: list[bytes] = [] + async with client.stream("GET", "/stream") as response: + async for chunk in response.aiter_bytes(): + chunks.append(chunk) + + assert b"".join(chunks) == b"firstsecond" + + async def test_request_body_and_headers_reach_the_app(self): + async def echo(request: Request) -> Response: + body = await request.body() + return Response( + content=body, + headers={"x-seen-header": request.headers.get("x-demo", "missing")}, + ) + + app = Starlette(routes=[Route("/echo", echo, methods=["POST"])]) + async with httpx2.AsyncClient( + transport=StreamingASGITransport(app), base_url="http://testserver" + ) as client: + response = await client.post( + "/echo", content=b"payload", headers={"x-demo": "abc"} + ) + + assert response.content == b"payload" + assert response.headers["x-seen-header"] == "abc" + + async def test_query_string_reaches_the_app(self): + async def show(request: Request) -> Response: + return Response(content=request.query_params["q"]) + + app = Starlette(routes=[Route("/search", show)]) + async with httpx2.AsyncClient( + transport=StreamingASGITransport(app), base_url="http://testserver" + ) as client: + response = await client.get("/search", params={"q": "hello"}) + + assert response.text == "hello" + + async def test_error_before_response_start_propagates_to_caller(self): + async def boom(scope: Scope, receive: Receive, send: Send) -> None: + raise ValueError("app exploded") + + async with httpx2.AsyncClient( + transport=StreamingASGITransport(boom), base_url="http://testserver" + ) as client: + with pytest.raises(ValueError, match="app exploded"): + await client.get("/anything") + + async def test_error_after_response_start_truncates_the_body(self): + """Post-start failures look like a dropped socket, not a raised exception.""" + + async def boom(scope: Scope, receive: Receive, send: Send) -> None: + await send( + { + "type": "http.response.start", + "status": 200, + "headers": [(b"content-type", b"text/plain")], + } + ) + # Raise with no checkpoint in between, so the error is guaranteed to be + # recorded before the transport's waiter resumes — the scheduling order + # that used to surface the failure as a raised exception. + raise ValueError("app exploded mid-response") + + async with httpx2.AsyncClient( + transport=StreamingASGITransport(boom), base_url="http://testserver" + ) as client: + response = await client.get("/anything") + + assert response.status_code == 200 + assert response.content == b"" + + +class TestRunAsgiLifespan: + async def test_startup_and_shutdown_run_once(self): + events: list[str] = [] + + async def app(scope: Scope, receive: Receive, send: Send) -> None: + assert scope["type"] == "lifespan" + while True: + message = await receive() + if message["type"] == "lifespan.startup": + events.append("startup") + await send({"type": "lifespan.startup.complete"}) + elif message["type"] == "lifespan.shutdown": + events.append("shutdown") + await send({"type": "lifespan.shutdown.complete"}) + return + + async with run_asgi_lifespan(app): + assert events == ["startup"] + + assert events == ["startup", "shutdown"] + + async def test_startup_failure_raises(self): + async def app(scope: Scope, receive: Receive, send: Send) -> None: + await receive() + await send({"type": "lifespan.startup.failed", "message": "nope"}) + + with pytest.raises(RuntimeError, match="startup failed"): + async with run_asgi_lifespan(app): + pass + + async def test_shutdown_failure_raises(self): + async def app(scope: Scope, receive: Receive, send: Send) -> None: + await receive() + await send({"type": "lifespan.startup.complete"}) + await receive() + await send({"type": "lifespan.shutdown.failed", "message": "teardown nope"}) + + with pytest.raises(RuntimeError, match="shutdown failed"): + async with run_asgi_lifespan(app): + pass + + async def test_shutdown_failure_does_not_mask_body_error(self): + """A broken teardown must never hide the failure the caller actually cares about.""" + + async def app(scope: Scope, receive: Receive, send: Send) -> None: + await receive() + await send({"type": "lifespan.startup.complete"}) + await receive() + await send({"type": "lifespan.shutdown.failed", "message": "teardown nope"}) + + with pytest.raises(ValueError, match="body exploded"): + async with run_asgi_lifespan(app): + raise ValueError("body exploded") + + +class TestRunServerInMemory: + @pytest.mark.parametrize("transport", ["http", "streamable-http", "sse"]) + async def test_client_round_trip( + self, transport: Literal["http", "streamable-http", "sse"] + ): + async with asgi_server(build_server(), transport=transport) as server: + async with server.client() as client: + result = await client.call_tool("greet", {"name": "World"}) + + assert result.data == "Hello, World!" + + async def test_default_path_matches_transport(self): + async with asgi_server(build_server(), transport="sse") as server: + assert server.url.endswith("/sse") + async with asgi_server(build_server(), transport="http") as server: + assert server.url.endswith("/mcp") + + async def test_custom_path_is_used(self): + async with asgi_server(build_server(), path="/custom") as server: + assert server.url.endswith("/custom") + async with server.client() as client: + assert await client.ping() is True + + async def test_server_initiated_request_mid_stream(self): + """Elicitation needs a server->client request while the POST is still open. + + This is the capability a buffering ASGI transport cannot provide, and the + reason the bridge streams responses. + """ + + async def elicitation_handler(message, response_type, params, ctx): + return {"value": "Alice"} + + async with asgi_server(build_server()) as server: + async with server.client(elicitation_handler=elicitation_handler) as client: + result = await client.call_tool("elicit_name", {}) + + assert result.data == "You said Alice" + + async def test_http_client_reaches_the_app(self): + async with asgi_server(build_server()) as server: + async with server.http_client() as http: + # A GET on the streamable HTTP endpoint without a session is rejected; + # the point is that the request reaches the real app at all. + response = await http.get(server.url) + + assert response.status_code in {400, 405} + + async def test_auth_middleware_runs(self): + """A migrated test must not pass by bypassing the real middleware stack.""" + key_pair = RSAKeyPair.generate() + server = build_server() + server.auth = JWTVerifier( + public_key=key_pair.public_key, + issuer="https://issuer.example.com", + audience="test-audience", + ) + + async with asgi_server(server) as running_server: + async with running_server.http_client() as http: + unauthenticated = await http.post( + running_server.url, + json={"jsonrpc": "2.0", "id": 1, "method": "initialize"}, + ) + + assert unauthenticated.status_code == 401 + + async def test_yields_in_memory_server(self): + async with asgi_server(build_server()) as server: + assert isinstance(server, ASGIServer) + assert server.transport_type == "http"