o i@sldZddlmZddlmZmZmZddlmZddl m Z er&ddl m Z edej dZGd d d ZdS) a Experimental client-side task support. This module provides client methods for interacting with MCP tasks. WARNING: These APIs are experimental and may change without notice. Example: # Call a tool as a task result = await session.experimental.call_tool_as_task("tool_name", {"arg": "value"}) task_id = result.task.taskId # Get task status status = await session.experimental.get_task(task_id) # Get task result when complete if status.status == "completed": result = await session.experimental.get_task_result(task_id, CallToolResult) # List all tasks tasks = await session.experimental.list_tasks() # Cancel a task await session.experimental.cancel_task(task_id) ) AsyncIterator) TYPE_CHECKINGAnyTypeVarN)poll_until_terminal) ClientSessionResultT)boundc @seZdZdZdddZ dddd d ed eeefdBd ed eeefdBde j f ddZ dede j fddZ dedeedefddZ ddedBde jfddZdede jfddZdedee j fddZdS)ExperimentalClientFeaturesz 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) sessionrreturnNcCs ||_dSN)_session)selfr rQ/home/ubuntu/.local/lib/python3.10/site-packages/mcp/client/experimental/tasks.py__init__1s z#ExperimentalClientFeatures.__init__i`)ttlmetaname argumentsrrc sZd}|durtjjdi|}|jttjtj||tj|d|ddtj IdHS)aCall 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: # Create task result = await session.experimental.call_tool_as_task( "long_running_tool", {"input": "data"} ) task_id = result.task.taskId # Poll for completion while True: status = await session.experimental.get_task(task_id) if status.status == "completed": break await asyncio.sleep(0.5) # Get result final = await session.experimental.get_task_result(task_id, CallToolResult) N)r)rrtask_metaparamsr) types RequestParamsMetar send_request ClientRequestCallToolRequestCallToolRequestParams TaskMetadataCreateTaskResult)rrrrrrrrrcall_tool_as_task4s"(  z,ExperimentalClientFeatures.call_tool_as_tasktask_idc.|jttjtj|ddtjIdHS)z Get the current status of a task. Args: task_id: The task identifier Returns: GetTaskResult containing the task status and metadata taskIdrN)rrrrGetTaskRequestGetTaskRequestParams GetTaskResultrr%rrrget_taskn   z#ExperimentalClientFeatures.get_task result_typecs,|jttjtj|dd|IdHS)a 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 r'rN)rrrrGetTaskPayloadRequestGetTaskPayloadRequestParams)rr%r/rrrget_task_results  z*ExperimentalClientFeatures.get_task_resultcursorcs:|r tj|dnd}|jttj|dtjIdHS)z List all tasks. Args: cursor: Optional pagination cursor Returns: ListTasksResult containing tasks and optional next cursor )r3Nr)rPaginatedRequestParamsrrrListTasksRequestListTasksResult)rr3rrrr list_taskss   z%ExperimentalClientFeatures.list_taskscr&)z Cancel a running task. Args: task_id: The task identifier Returns: CancelTaskResult with the updated task state r'rN)rrrrCancelTaskRequestCancelTaskRequestParamsCancelTaskResultr,rrr cancel_taskr.z&ExperimentalClientFeatures.cancel_taskcCs(t|j|2z 3dHW}|Vq6dS)ak 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 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: 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) N)rr-)rr%statusrrr poll_tasksz$ExperimentalClientFeatures.poll_task)r rr Nr )__name__ __module__ __qualname____doc__rstrdictrintrr#r$r+r-typerr2r6r7r:r;rr=rrrrr 'sD  :  r )rAcollections.abcrtypingrrr mcp.typesr%mcp.shared.experimental.tasks.pollingrmcp.client.sessionrResultrr rrrrs