fastmcp.utilities.async_utils

Async utilities for FastMCP.

Functions

is_coroutine_function ^{<a href="https://github.com/PrefectHQ/fastmcp/blob/main/src/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/src/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/src/fastmcp/utilities/async_utils.py#L51" target="_blank"><Icon icon="github" style="width: 14px; height: 14px;" /></a>}

gather(*awaitables: Awaitable[T]) -> list[T] | list[T | BaseException]

Run awaitables concurrently and return results in order.

Uses anyio TaskGroup for structured concurrency.

Args:

• *awaitables: 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.