fastmcp.utilities.async_utils

Async utilities for FastMCP.

Functions

is_coroutine_function ^{<a href="https://github.com/PrefectHQ/fastmcp/blob/main/fastmcp_slim/fastmcp/utilities/async_utils.py#L14" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

is_coroutine_function(fn: Any) -> bool

Check if a callable is a coroutine function, unwrapping functools.partial.

inspect.iscoroutinefunction returns False for functools.partial objects wrapping an async function on Python < 3.12. This helper unwraps any layers of partial before checking.

call_sync_fn_in_threadpool ^{<a href="https://github.com/PrefectHQ/fastmcp/blob/main/fastmcp_slim/fastmcp/utilities/async_utils.py#L26" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

call_sync_fn_in_threadpool(fn: Callable[..., Any], *args: Any, **kwargs: Any) -> Any

Call a sync function in a threadpool to avoid blocking the event loop.

Uses anyio.to_thread.run_sync which properly propagates contextvars, making this safe for functions that depend on context (like dependency injection).

gather ^{<a href="https://github.com/PrefectHQ/fastmcp/blob/main/fastmcp_slim/fastmcp/utilities/async_utils.py#L53" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

gather(awaitables: Iterable[Awaitable[T]]) -> 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: 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.