Skip to content

Index

Experimental client features.

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

ExperimentalClientFeatures

Experimental client features for tasks and other experimental APIs.

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

Access via session.experimental: status = await session.experimental.get_task(task_id)

Source code in src/mcp/client/experimental/tasks.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class ExperimentalClientFeatures:
    """Experimental client features for tasks and other experimental APIs.

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

    Access via session.experimental:
        status = await session.experimental.get_task(task_id)
    """

    def __init__(self, session: "ClientSession") -> None:
        self._session = session

    async def call_tool_as_task(
        self,
        name: str,
        arguments: dict[str, Any] | None = None,
        *,
        ttl: int = 60000,
        meta: RequestParamsMeta | None = None,
    ) -> types.CreateTaskResult:
        """Call a tool as a task, returning a CreateTaskResult for polling.

        This is a convenience method for calling tools that support task execution.
        The server will return a task reference instead of the immediate result,
        which can then be polled via `get_task()` and retrieved via `get_task_result()`.

        Args:
            name: The tool name
            arguments: Tool arguments
            ttl: Task time-to-live in milliseconds (default: 60000 = 1 minute)
            meta: Optional metadata to include in the request

        Returns:
            CreateTaskResult containing the task reference

        Example:
            ```python
            # Create task
            result = await session.experimental.call_tool_as_task(
                "long_running_tool", {"input": "data"}
            )
            task_id = result.task.task_id

            # Poll for completion
            while True:
                status = await session.experimental.get_task(task_id)
                if status.status == "completed":
                    break
                await anyio.sleep(0.5)

            # Get result
            final = await session.experimental.get_task_result(task_id, CallToolResult)
            ```
        """
        return await self._session.send_request(
            types.CallToolRequest(
                params=types.CallToolRequestParams(
                    name=name,
                    arguments=arguments,
                    task=types.TaskMetadata(ttl=ttl),
                    _meta=meta,
                ),
            ),
            types.CreateTaskResult,
        )

    async def get_task(self, task_id: str) -> types.GetTaskResult:
        """Get the current status of a task.

        Args:
            task_id: The task identifier

        Returns:
            GetTaskResult containing the task status and metadata
        """
        return await self._session.send_request(
            types.GetTaskRequest(params=types.GetTaskRequestParams(task_id=task_id)),
            types.GetTaskResult,
        )

    async def get_task_result(
        self,
        task_id: str,
        result_type: type[ResultT],
    ) -> ResultT:
        """Get the result of a completed task.

        The result type depends on the original request type:
        - tools/call tasks return CallToolResult
        - Other request types return their corresponding result type

        Args:
            task_id: The task identifier
            result_type: The expected result type (e.g., CallToolResult)

        Returns:
            The task result, validated against result_type
        """
        return await self._session.send_request(
            types.GetTaskPayloadRequest(
                params=types.GetTaskPayloadRequestParams(task_id=task_id),
            ),
            result_type,
        )

    async def list_tasks(
        self,
        cursor: str | None = None,
    ) -> types.ListTasksResult:
        """List all tasks.

        Args:
            cursor: Optional pagination cursor

        Returns:
            ListTasksResult containing tasks and optional next cursor
        """
        params = types.PaginatedRequestParams(cursor=cursor) if cursor else None
        return await self._session.send_request(
            types.ListTasksRequest(params=params),
            types.ListTasksResult,
        )

    async def cancel_task(self, task_id: str) -> types.CancelTaskResult:
        """Cancel a running task.

        Args:
            task_id: The task identifier

        Returns:
            CancelTaskResult with the updated task state
        """
        return await self._session.send_request(
            types.CancelTaskRequest(
                params=types.CancelTaskRequestParams(task_id=task_id),
            ),
            types.CancelTaskResult,
        )

    async def poll_task(self, task_id: str) -> AsyncIterator[types.GetTaskResult]:
        """Poll a task until it reaches a terminal status.

        Yields GetTaskResult for each poll, allowing the caller to react to
        status changes (e.g., handle input_required). Exits when the task reaches
        a terminal status (completed, failed, cancelled).

        Respects the pollInterval hint from the server.

        Args:
            task_id: The task identifier

        Yields:
            GetTaskResult for each poll

        Example:
            ```python
            async for status in session.experimental.poll_task(task_id):
                print(f"Status: {status.status}")
                if status.status == "input_required":
                    # Handle elicitation request via tasks/result
                    pass

            # Task is now terminal, get the result
            result = await session.experimental.get_task_result(task_id, CallToolResult)
            ```
        """
        async for status in poll_until_terminal(self.get_task, task_id):
            yield status

call_tool_as_task async 

call_tool_as_task(
    name: str,
    arguments: dict[str, Any] | None = None,
    *,
    ttl: int = 60000,
    meta: RequestParamsMeta | None = None
) -> CreateTaskResult

Call a tool as a task, returning a CreateTaskResult for polling.

This is a convenience method for calling tools that support task execution. The server will return a task reference instead of the immediate result, which can then be polled via get_task() and retrieved via get_task_result().

Parameters:

Name Type Description Default
name str

The tool name

 required
arguments dict[str, Any] | None

Tool arguments

 None
ttl int

Task time-to-live in milliseconds (default: 60000 = 1 minute)

 60000
meta RequestParamsMeta | None

Optional metadata to include in the request

 None

Returns:

Type Description
CreateTaskResult

CreateTaskResult containing the task reference
Example 
# Create task
result = await session.experimental.call_tool_as_task(
    "long_running_tool", {"input": "data"}
)
task_id = result.task.task_id

# Poll for completion
while True:
    status = await session.experimental.get_task(task_id)
    if status.status == "completed":
        break
    await anyio.sleep(0.5)

# Get result
final = await session.experimental.get_task_result(task_id, CallToolResult)
Source code in src/mcp/client/experimental/tasks.py 
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
async def call_tool_as_task(
    self,
    name: str,
    arguments: dict[str, Any] | None = None,
    *,
    ttl: int = 60000,
    meta: RequestParamsMeta | None = None,
) -> types.CreateTaskResult:
    """Call a tool as a task, returning a CreateTaskResult for polling.

    This is a convenience method for calling tools that support task execution.
    The server will return a task reference instead of the immediate result,
    which can then be polled via `get_task()` and retrieved via `get_task_result()`.

    Args:
        name: The tool name
        arguments: Tool arguments
        ttl: Task time-to-live in milliseconds (default: 60000 = 1 minute)
        meta: Optional metadata to include in the request

    Returns:
        CreateTaskResult containing the task reference

    Example:
        ```python
        # Create task
        result = await session.experimental.call_tool_as_task(
            "long_running_tool", {"input": "data"}
        )
        task_id = result.task.task_id

        # Poll for completion
        while True:
            status = await session.experimental.get_task(task_id)
            if status.status == "completed":
                break
            await anyio.sleep(0.5)

        # Get result
        final = await session.experimental.get_task_result(task_id, CallToolResult)
        ```
    """
    return await self._session.send_request(
        types.CallToolRequest(
            params=types.CallToolRequestParams(
                name=name,
                arguments=arguments,
                task=types.TaskMetadata(ttl=ttl),
                _meta=meta,
            ),
        ),
        types.CreateTaskResult,
    )

get_task async 

get_task(task_id: str) -> GetTaskResult

Get the current status of a task.

Parameters:

Name Type Description Default
task_id str

The task identifier

 required

Returns:

Type Description
GetTaskResult

GetTaskResult containing the task status and metadata
Source code in src/mcp/client/experimental/tasks.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
async def get_task(self, task_id: str) -> types.GetTaskResult:
    """Get the current status of a task.

    Args:
        task_id: The task identifier

    Returns:
        GetTaskResult containing the task status and metadata
    """
    return await self._session.send_request(
        types.GetTaskRequest(params=types.GetTaskRequestParams(task_id=task_id)),
        types.GetTaskResult,
    )

get_task_result async 

get_task_result(
    task_id: str, result_type: type[ResultT]
) -> ResultT

Get the result of a completed task.

The result type depends on the original request type: - tools/call tasks return CallToolResult - Other request types return their corresponding result type

Parameters:

Name Type Description Default
task_id str

The task identifier

 required
result_type type[ResultT]

The expected result type (e.g., CallToolResult)

 required

Returns:

Type Description
ResultT

The task result, validated against result_type
Source code in src/mcp/client/experimental/tasks.py 
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
async def get_task_result(
    self,
    task_id: str,
    result_type: type[ResultT],
) -> ResultT:
    """Get the result of a completed task.

    The result type depends on the original request type:
    - tools/call tasks return CallToolResult
    - Other request types return their corresponding result type

    Args:
        task_id: The task identifier
        result_type: The expected result type (e.g., CallToolResult)

    Returns:
        The task result, validated against result_type
    """
    return await self._session.send_request(
        types.GetTaskPayloadRequest(
            params=types.GetTaskPayloadRequestParams(task_id=task_id),
        ),
        result_type,
    )

list_tasks async 

list_tasks(cursor: str | None = None) -> ListTasksResult

List all tasks.

Parameters:

Name Type Description Default
cursor str | None

Optional pagination cursor

 None

Returns:

Type Description
ListTasksResult

ListTasksResult containing tasks and optional next cursor
Source code in src/mcp/client/experimental/tasks.py 
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
async def list_tasks(
    self,
    cursor: str | None = None,
) -> types.ListTasksResult:
    """List all tasks.

    Args:
        cursor: Optional pagination cursor

    Returns:
        ListTasksResult containing tasks and optional next cursor
    """
    params = types.PaginatedRequestParams(cursor=cursor) if cursor else None
    return await self._session.send_request(
        types.ListTasksRequest(params=params),
        types.ListTasksResult,
    )

cancel_task async 

cancel_task(task_id: str) -> CancelTaskResult

Cancel a running task.

Parameters:

Name Type Description Default
task_id str

The task identifier

 required

Returns:

Type Description
CancelTaskResult

CancelTaskResult with the updated task state
Source code in src/mcp/client/experimental/tasks.py 
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
async def cancel_task(self, task_id: str) -> types.CancelTaskResult:
    """Cancel a running task.

    Args:
        task_id: The task identifier

    Returns:
        CancelTaskResult with the updated task state
    """
    return await self._session.send_request(
        types.CancelTaskRequest(
            params=types.CancelTaskRequestParams(task_id=task_id),
        ),
        types.CancelTaskResult,
    )

poll_task async 

poll_task(task_id: str) -> AsyncIterator[GetTaskResult]

Poll a task until it reaches a terminal status.

Yields GetTaskResult for each poll, allowing the caller to react to status changes (e.g., handle input_required). Exits when the task reaches a terminal status (completed, failed, cancelled).

Respects the pollInterval hint from the server.

Parameters:

Name Type Description Default
task_id str

The task identifier

 required

Yields:

Type Description
AsyncIterator[GetTaskResult]

GetTaskResult for each poll
Example 
async for status in session.experimental.poll_task(task_id):
    print(f"Status: {status.status}")
    if status.status == "input_required":
        # Handle elicitation request via tasks/result
        pass

# Task is now terminal, get the result
result = await session.experimental.get_task_result(task_id, CallToolResult)
Source code in src/mcp/client/experimental/tasks.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
async def poll_task(self, task_id: str) -> AsyncIterator[types.GetTaskResult]:
    """Poll a task until it reaches a terminal status.

    Yields GetTaskResult for each poll, allowing the caller to react to
    status changes (e.g., handle input_required). Exits when the task reaches
    a terminal status (completed, failed, cancelled).

    Respects the pollInterval hint from the server.

    Args:
        task_id: The task identifier

    Yields:
        GetTaskResult for each poll

    Example:
        ```python
        async for status in session.experimental.poll_task(task_id):
            print(f"Status: {status.status}")
            if status.status == "input_required":
                # Handle elicitation request via tasks/result
                pass

        # Task is now terminal, get the result
        result = await session.experimental.get_task_result(task_id, CallToolResult)
        ```
    """
    async for status in poll_until_terminal(self.get_task, task_id):
        yield status