polling

Shared polling utilities for task operations.

This module provides generic polling logic that works for both client→server and server→client task polling.

WARNING: These APIs are experimental and may change without notice.

poll_until_terminal async

poll_until_terminal(
    get_task: Callable[[str], Awaitable[GetTaskResult]],
    task_id: str,
    default_interval_ms: int = 500,
) -> AsyncIterator[GetTaskResult]

Poll a task until it reaches terminal status.

This is a generic utility that works for both client→server and server→client polling. The caller provides the get_task function appropriate for their direction.

Parameters:

Name	Type	Description	Default
get_task	Callable[[str], Awaitable[GetTaskResult]]	Async function that takes task_id and returns GetTaskResult	required
task_id	str	The task to poll	required
default_interval_ms	int	Fallback poll interval if server doesn't specify	500

Yields:

Type	Description
AsyncIterator[GetTaskResult]	GetTaskResult for each poll

Source code in src/mcp/shared/experimental/tasks/polling.py

async def poll_until_terminal(
    get_task: Callable[[str], Awaitable[GetTaskResult]],
    task_id: str,
    default_interval_ms: int = 500,
) -> AsyncIterator[GetTaskResult]:
    """Poll a task until it reaches terminal status.

    This is a generic utility that works for both client→server and server→client
    polling. The caller provides the get_task function appropriate for their direction.

    Args:
        get_task: Async function that takes task_id and returns GetTaskResult
        task_id: The task to poll
        default_interval_ms: Fallback poll interval if server doesn't specify

    Yields:
        GetTaskResult for each poll
    """
    while True:
        status = await get_task(task_id)
        yield status

        if is_terminal(status.status):
            break

        interval_ms = status.poll_interval if status.poll_interval is not None else default_interval_ms
        await anyio.sleep(interval_ms / 1000)