prefab-ui not installed. Run: pip install fastmcp[apps]
" + return "prefab-ui not installed. Run: pip install 'fastmcp[apps]'
" if not tools: with Column(gap=4, css_class="p-6 max-w-2xl mx-auto") as view: diff --git a/fastmcp_slim/fastmcp/client/sampling/handlers/anthropic.py b/fastmcp_slim/fastmcp/client/sampling/handlers/anthropic.py index 88eb132ee..747d1ab69 100644 --- a/fastmcp_slim/fastmcp/client/sampling/handlers/anthropic.py +++ b/fastmcp_slim/fastmcp/client/sampling/handlers/anthropic.py @@ -41,7 +41,7 @@ try: except ImportError as e: raise ImportError( "The `anthropic` package is not installed. " - "Install it with `pip install fastmcp-slim[anthropic]` or add `anthropic` to your dependencies." + "Install it with `pip install 'fastmcp-slim[anthropic]'` or add `anthropic` to your dependencies." ) from e __all__ = ["AnthropicSamplingHandler"] diff --git a/fastmcp_slim/fastmcp/client/sampling/handlers/google_genai.py b/fastmcp_slim/fastmcp/client/sampling/handlers/google_genai.py index 12d6064e6..7bfbbb626 100644 --- a/fastmcp_slim/fastmcp/client/sampling/handlers/google_genai.py +++ b/fastmcp_slim/fastmcp/client/sampling/handlers/google_genai.py @@ -28,7 +28,7 @@ try: except ImportError as e: raise ImportError( "The `google-genai` package is not installed. " - "Install it with `pip install fastmcp-slim[gemini]` or add `google-genai` " + "Install it with `pip install 'fastmcp-slim[gemini]'` or add `google-genai` " "to your dependencies." ) from e diff --git a/fastmcp_slim/fastmcp/server/providers/aggregate.py b/fastmcp_slim/fastmcp/server/providers/aggregate.py index 766d64846..53928d055 100644 --- a/fastmcp_slim/fastmcp/server/providers/aggregate.py +++ b/fastmcp_slim/fastmcp/server/providers/aggregate.py @@ -190,7 +190,7 @@ class AggregateProvider(Provider): async def _list_tools(self) -> Sequence[Tool]: """List all tools from all providers.""" results = await gather( - *[p.list_tools() for p in self.providers], + (p.list_tools() for p in self.providers), return_exceptions=True, ) return self._collect_list_results(results, "list_tools") @@ -200,7 +200,7 @@ class AggregateProvider(Provider): ) -> Tool | None: """Get tool by name from providers.""" results = await gather( - *[p.get_tool(name, version) for p in self.providers], + (p.get_tool(name, version) for p in self.providers), return_exceptions=True, ) return self._get_highest_version_result(results, f"get_tool({name!r})") # type: ignore[return-value] # ty:ignore[invalid-argument-type, invalid-return-type] @@ -208,7 +208,7 @@ class AggregateProvider(Provider): async def get_app_tool(self, app_name: str, tool_name: str) -> Tool | None: """Query all child providers for an app tool.""" results = await gather( - *[p.get_app_tool(app_name, tool_name) for p in self.providers], + (p.get_app_tool(app_name, tool_name) for p in self.providers), return_exceptions=True, ) for r in results: @@ -223,7 +223,7 @@ class AggregateProvider(Provider): async def get_tool_by_hash(self, tool_hash: str, tool_name: str) -> Tool | None: """Query all child providers for a tool matching a hash.""" results = await gather( - *[p.get_tool_by_hash(tool_hash, tool_name) for p in self.providers], + (p.get_tool_by_hash(tool_hash, tool_name) for p in self.providers), return_exceptions=True, ) for r in results: @@ -242,7 +242,7 @@ class AggregateProvider(Provider): async def _list_resources(self) -> Sequence[Resource]: """List all resources from all providers.""" results = await gather( - *[p.list_resources() for p in self.providers], + (p.list_resources() for p in self.providers), return_exceptions=True, ) return self._collect_list_results(results, "list_resources") @@ -252,7 +252,7 @@ class AggregateProvider(Provider): ) -> Resource | None: """Get resource by URI from providers.""" results = await gather( - *[p.get_resource(uri, version) for p in self.providers], + (p.get_resource(uri, version) for p in self.providers), return_exceptions=True, ) return self._get_highest_version_result(results, f"get_resource({uri!r})") # type: ignore[return-value] # ty:ignore[invalid-argument-type, invalid-return-type] @@ -264,7 +264,7 @@ class AggregateProvider(Provider): async def _list_resource_templates(self) -> Sequence[ResourceTemplate]: """List all resource templates from all providers.""" results = await gather( - *[p.list_resource_templates() for p in self.providers], + (p.list_resource_templates() for p in self.providers), return_exceptions=True, ) return self._collect_list_results(results, "list_resource_templates") @@ -274,7 +274,7 @@ class AggregateProvider(Provider): ) -> ResourceTemplate | None: """Get resource template by URI from providers.""" results = await gather( - *[p.get_resource_template(uri, version) for p in self.providers], + (p.get_resource_template(uri, version) for p in self.providers), return_exceptions=True, ) return self._get_highest_version_result( @@ -288,7 +288,7 @@ class AggregateProvider(Provider): async def _list_prompts(self) -> Sequence[Prompt]: """List all prompts from all providers.""" results = await gather( - *[p.list_prompts() for p in self.providers], + (p.list_prompts() for p in self.providers), return_exceptions=True, ) return self._collect_list_results(results, "list_prompts") @@ -298,7 +298,7 @@ class AggregateProvider(Provider): ) -> Prompt | None: """Get prompt by name from providers.""" results = await gather( - *[p.get_prompt(name, version) for p in self.providers], + (p.get_prompt(name, version) for p in self.providers), return_exceptions=True, ) return self._get_highest_version_result(results, f"get_prompt({name!r})") # type: ignore[return-value] # ty:ignore[invalid-argument-type, invalid-return-type] @@ -310,7 +310,7 @@ class AggregateProvider(Provider): async def get_tasks(self) -> Sequence[FastMCPComponent]: """Get all task-eligible components from all providers.""" results = await gather( - *[p.get_tasks() for p in self.providers], + (p.get_tasks() for p in self.providers), return_exceptions=True, ) return self._collect_list_results(results, "get_tasks") diff --git a/fastmcp_slim/fastmcp/server/providers/base.py b/fastmcp_slim/fastmcp/server/providers/base.py index 402dff351..959dc9de9 100644 --- a/fastmcp_slim/fastmcp/server/providers/base.py +++ b/fastmcp_slim/fastmcp/server/providers/base.py @@ -496,12 +496,19 @@ class Provider: Used by the server during startup to register functions with Docket. """ - # Fetch all component types in parallel + # Fetch all component types in parallel. Iterate the bound methods + # rather than a tuple of already-called coroutines: a parenthesized + # comma expression is a tuple, so it would create all four coroutines + # before `gather` starts, which is exactly what `gather` asks callers + # to avoid. results = await gather( - self._list_tools(), - self._list_resources(), - self._list_resource_templates(), - self._list_prompts(), + fetch() + for fetch in ( + self._list_tools, + self._list_resources, + self._list_resource_templates, + self._list_prompts, + ) ) tools = cast("Sequence[Tool]", results[0]) resources = cast("Sequence[Resource]", results[1]) diff --git a/fastmcp_slim/fastmcp/server/sampling/run.py b/fastmcp_slim/fastmcp/server/sampling/run.py index bcb73cc66..4aedd60ec 100644 --- a/fastmcp_slim/fastmcp/server/sampling/run.py +++ b/fastmcp_slim/fastmcp/server/sampling/run.py @@ -390,7 +390,7 @@ async def execute_tools( # Execute in parallel if tool_concurrency == 0: # Unlimited parallel execution - return await gather(*[_execute_single_tool(tc) for tc in tool_calls]) + return await gather(_execute_single_tool(tc) for tc in tool_calls) else: # Bounded parallel execution with semaphore semaphore = anyio.Semaphore(tool_concurrency) @@ -399,7 +399,7 @@ async def execute_tools( async with semaphore: return await _execute_single_tool(tool_use) - return await gather(*[bounded_execute(tc) for tc in tool_calls]) + return await gather(bounded_execute(tc) for tc in tool_calls) # --- Helper functions for sampling --- diff --git a/fastmcp_slim/fastmcp/tools/tool_transform.py b/fastmcp_slim/fastmcp/tools/tool_transform.py index 910b89262..effb5386b 100644 --- a/fastmcp_slim/fastmcp/tools/tool_transform.py +++ b/fastmcp_slim/fastmcp/tools/tool_transform.py @@ -707,7 +707,8 @@ class TransformedTool(Tool): schema = { "type": "object", "properties": new_props, - "required": list(new_required), + # Iterate props (not the set) for deterministic ordering + "required": [p for p in new_props if p in new_required], "additionalProperties": False, } @@ -899,7 +900,11 @@ class TransformedTool(Tool): result = { "type": "object", "properties": merged_props, - "required": list(final_required), + # Iterate props (not the set) for deterministic ordering; keep any + # required names not present in properties (sorted) rather than + # silently dropping them. + "required": [p for p in merged_props if p in final_required] + + sorted(final_required - set(merged_props)), "additionalProperties": False, } 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/async_utils.py b/fastmcp_slim/fastmcp/utilities/async_utils.py index 8fc24f49d..978c84a6e 100644 --- a/fastmcp_slim/fastmcp/utilities/async_utils.py +++ b/fastmcp_slim/fastmcp/utilities/async_utils.py @@ -2,7 +2,7 @@ import functools import inspect -from collections.abc import Awaitable, Callable +from collections.abc import Awaitable, Callable, Iterable from typing import Any, Literal, TypeVar, overload import anyio @@ -36,35 +36,55 @@ async def call_sync_fn_in_threadpool( @overload async def gather( - *awaitables: Awaitable[T], + awaitables: Iterable[Awaitable[T]], + *, return_exceptions: Literal[True], ) -> list[T | BaseException]: ... @overload async def gather( - *awaitables: Awaitable[T], + awaitables: Iterable[Awaitable[T]], + *, return_exceptions: Literal[False] = ..., ) -> list[T]: ... async def gather( - *awaitables: Awaitable[T], + awaitables: Iterable[Awaitable[T]], + *, return_exceptions: bool = False, ) -> list[T] | list[T | BaseException]: """Run awaitables concurrently and return results in order. Uses anyio TaskGroup for structured concurrency. + ``awaitables`` is consumed lazily, one item at a time, right before each + is handed to the task group. Callers with a dynamic number of awaitables + should pass a generator expression (e.g. ``gather(f(x) for x in xs)``) + rather than a list or list comprehension: a list comprehension calls + every ``f(x)`` up front, creating a batch of coroutine objects before + this function even starts, whereas a generator expression creates each + coroutine only as this function's own scheduling loop asks for it. That + matters because coroutine creation and scheduling can be interrupted + between any two bytecode instructions by a synchronous signal handler + (for example pytest-timeout's SIGALRM-based per-test timeout). If that + happens while a whole batch of coroutines is sitting unscheduled, they + are silently abandoned and eventually trigger a "coroutine was never + awaited" warning attributed to whatever unrelated code happens to be + running when the garbage collector gets to them. Lazy consumption keeps + the window in which a created-but-unscheduled coroutine can exist as + small as possible. + Args: - *awaitables: Awaitables to run concurrently + awaitables: Iterable of awaitables to run concurrently. return_exceptions: If True, exceptions are returned in results. If False, first exception cancels all and raises. Returns: List of results in the same order as input awaitables. """ - results: list[T | BaseException] = [None] * len(awaitables) # type: ignore[assignment] # ty:ignore[invalid-assignment] + results: list[T | BaseException] = [] async def run_at(i: int, aw: Awaitable[T]) -> None: try: @@ -75,8 +95,26 @@ async def gather( else: raise + pending = enumerate(awaitables) async with anyio.create_task_group() as tg: - for i, aw in enumerate(awaitables): - tg.start_soon(run_at, i, aw) + for i, aw in pending: + results.append(None) # type: ignore[arg-type] # ty:ignore[invalid-argument-type] + try: + tg.start_soon(run_at, i, aw) + except BaseException: + # `aw` was just created (possibly moments ago, by the + # generator's own iteration) but never handed off - close it + # explicitly so it isn't silently garbage collected later. + if inspect.iscoroutine(aw): + aw.close() + # Lazy consumption keeps the leak window small, but a caller + # that passed an already-built sequence has coroutines sitting + # behind this one that were never scheduled either. Draining + # the iterator closes them too, so `gather` cannot leak + # regardless of how eagerly its argument was constructed. + for _, remaining in pending: + if inspect.iscoroutine(remaining): + remaining.close() + raise return results 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/tools/tool_transform/test_tool_transform.py b/tests/tools/tool_transform/test_tool_transform.py index f9e50df72..2cde40b7f 100644 --- a/tests/tools/tool_transform/test_tool_transform.py +++ b/tests/tools/tool_transform/test_tool_transform.py @@ -41,6 +41,23 @@ def test_tool_from_tool_no_change(add_tool): assert new_tool.description == add_tool.description +def test_transformed_tool_required_order_is_deterministic(): + """`required` must follow property order, not set iteration order. + + Set iteration order varies with PYTHONHASHSEED, which broke snapshot + tests of tools/list output across processes. + """ + + def fn(alpha: int, beta: str, gamma: float, delta: bool, epsilon: int) -> str: + return "x" + + base = Tool.from_function(fn) + transformed = Tool.from_tool(base, transform_args={"alpha": ArgTransform(name="a")}) + props = list(transformed.parameters["properties"]) + assert transformed.parameters["required"] == props + assert props == ["a", "beta", "gamma", "delta", "epsilon"] + + def test_from_tool_accepts_decorated_function(): @tool def search(q: str, limit: int = 10) -> list[str]: 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" diff --git a/tests/utilities/test_async_utils.py b/tests/utilities/test_async_utils.py index 6d1366613..7f9d02d5d 100644 --- a/tests/utilities/test_async_utils.py +++ b/tests/utilities/test_async_utils.py @@ -1,7 +1,11 @@ """Tests for fastmcp.utilities.async_utils.""" import functools +import inspect +from collections.abc import Awaitable, Iterator +from typing import Any +import anyio import pytest from exceptiongroup import BaseExceptionGroup @@ -9,6 +13,7 @@ from fastmcp import Client, FastMCP from fastmcp.prompts import prompt from fastmcp.resources import resource from fastmcp.tools import tool +from fastmcp.utilities import async_utils from fastmcp.utilities.async_utils import gather, is_coroutine_function @@ -55,14 +60,20 @@ class TestGather: async def value(result: int) -> int: return result - assert await gather(value(1), value(2), value(3)) == [1, 2, 3] + assert await gather([value(1), value(2), value(3)]) == [1, 2, 3] + + async def test_accepts_a_generator(self) -> None: + async def value(result: int) -> int: + return result + + assert await gather(value(i) for i in [1, 2, 3]) == [1, 2, 3] async def test_raises_by_default(self) -> None: async def fail() -> int: raise RuntimeError("boom") with pytest.raises(BaseExceptionGroup) as exc_info: - await gather(fail()) + await gather([fail()]) assert len(exc_info.value.exceptions) == 1 assert isinstance(exc_info.value.exceptions[0], RuntimeError) @@ -74,11 +85,118 @@ class TestGather: async def value() -> int: return 1 - result = await gather(fail(), value(), return_exceptions=True) + result = await gather([fail(), value()], return_exceptions=True) assert isinstance(result[0], ValueError) assert result[1] == 1 + async def test_does_not_leak_coroutine_when_scheduling_is_interrupted( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """If handing an already-created awaitable off to the task group + raises partway through scheduling, that awaitable must be closed + rather than silently garbage collected later - which is what + produces a "coroutine was never awaited" RuntimeWarning attributed + to whatever unrelated code happens to be running when the garbage + collector eventually reclaims it. + + In production this can happen when a synchronous signal handler + (e.g. pytest-timeout's SIGALRM-based per-test timeout) fires inside + anyio's task-spawning internals. This test reproduces the same + shape of interruption deterministically by making the task group's + ``start_soon`` raise partway through scheduling, instead of relying + on real signal timing. + """ + real_create_task_group = anyio.create_task_group + + class _FailOnSecondStart: + def __init__(self) -> None: + self._real_tg = real_create_task_group() + self._calls = 0 + + async def __aenter__(self) -> "_FailOnSecondStart": + await self._real_tg.__aenter__() + return self + + async def __aexit__(self, *exc_info: Any) -> bool | None: + return await self._real_tg.__aexit__(*exc_info) + + def start_soon(self, func: Any, *args: Any) -> None: + self._calls += 1 + if self._calls == 2: + raise RuntimeError("interrupted while scheduling") + self._real_tg.start_soon(func, *args) + + monkeypatch.setattr(async_utils.anyio, "create_task_group", _FailOnSecondStart) + + created: list[Any] = [] + + async def value(result: int) -> int: + return result + + def awaitables() -> Iterator[Awaitable[int]]: + for i in range(3): + aw = value(i) + created.append(aw) + yield aw + + with pytest.raises(BaseExceptionGroup) as exc_info: + await gather(awaitables()) + + assert len(exc_info.value.exceptions) == 1 + assert isinstance(exc_info.value.exceptions[0], RuntimeError) + assert "interrupted while scheduling" in str(exc_info.value.exceptions[0]) + + # created[1] was being handed to start_soon() when it raised - it + # must have been closed rather than abandoned. + assert inspect.getcoroutinestate(created[1]) == "CORO_CLOSED" + + async def test_closes_unscheduled_coroutines_from_an_eager_caller( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + """Lazy consumption keeps the leak window small, but a caller that + builds its awaitables eagerly (a list or a parenthesized tuple) has + coroutines queued behind the failing one that were never scheduled + either. ``gather`` drains what is left of the iterable and closes + those too, so it cannot leak regardless of how its argument was + constructed.""" + real_create_task_group = anyio.create_task_group + + class _FailOnSecondStart: + def __init__(self) -> None: + self._real_tg = real_create_task_group() + self._calls = 0 + + async def __aenter__(self) -> "_FailOnSecondStart": + await self._real_tg.__aenter__() + return self + + async def __aexit__(self, *exc_info: Any) -> bool | None: + return await self._real_tg.__aexit__(*exc_info) + + def start_soon(self, func: Any, *args: Any) -> None: + self._calls += 1 + if self._calls == 2: + raise RuntimeError("interrupted while scheduling") + self._real_tg.start_soon(func, *args) + + monkeypatch.setattr(async_utils.anyio, "create_task_group", _FailOnSecondStart) + + async def value(result: int) -> int: + return result + + # Eagerly built: all four coroutines exist before gather() runs. + eager = [value(0), value(1), value(2), value(3)] + + with pytest.raises(BaseExceptionGroup): + await gather(eager) + + # The one that failed to schedule *and* the two queued behind it are + # all closed; none is left to surface as a stray warning later. + assert [inspect.getcoroutinestate(aw) for aw in eager[1:]] == [ + "CORO_CLOSED" + ] * 3 + class TestAsyncPartialIntegration: async def test_async_partial_tool_runs(self) -> None: