o c۷i!@sdZddlmZmZddlmZmZddlmZddl m Z ddl m Z ddl mZddlmZdd lmZmZdd lmZmZmZmZmZmZmZmZmZmZeGd d d Zd S)a  Experimental request context features. This module provides the Experimental class which gives access to experimental features within a request context, such as task-augmented request handling. WARNING: These APIs are experimental and may change without notice. ) AwaitableCallable) dataclassfield)Any)ServerTaskContext) TaskSupport) ServerSession)McpError)MODEL_IMMEDIATE_RESPONSE_KEY is_terminal) METHOD_NOT_FOUNDTASK_FORBIDDEN TASK_REQUIREDClientCapabilitiesCreateTaskResult ErrorDataResultTaskExecutionMode TaskMetadataToolc @s eZdZUdZdZedBed<edddZe dBed<edddZ e dBed<edddZ e dBed<ed efd d Zed efd d ZdddedBded edBfddZdddeded edBfddZdedBd efddZddddeegeefdedBdedBd efddZdS) Experimentala Experimental features context for task-augmented requests. Provides helpers for validating task execution compatibility and running tasks with automatic lifecycle management. WARNING: This API is experimental and may change without notice. N task_metadataF)defaultrepr_client_capabilities_session _task_supportreturncCs |jduS)z(Check if this request is task-augmented.N)rselfr!]/home/ubuntu/vllm_env/lib/python3.10/site-packages/mcp/server/experimental/request_context.pyis_task1s zExperimental.is_taskcCs|jdurdS|jjduS)z*Check if the client declared task support.NF)rtasksrr!r!r"client_supports_tasks6s  z"Experimental.client_supports_tasksT raise_errortool_task_moder'cCsZ|pt}d}|tkr|jsttdd}n |tkr!|jr!ttdd}|dur+|r+t||S)a Validate that the request is compatible with the tool's task execution mode. Per MCP spec: - "required": Clients MUST invoke as task. Server returns -32601 if not. - "forbidden" (or None): Clients MUST NOT invoke as task. Server returns -32601 if they do. - "optional": Either is acceptable. Args: tool_task_mode: The tool's execution.taskSupport value ("forbidden", "optional", "required", or None) raise_error: If True, raises McpError on validation failure. If False, returns ErrorData. Returns: None if valid, ErrorData if invalid and raise_error=False Raises: McpError: If invalid and raise_error=True Nz,This tool requires task-augmented invocation)codemessagez4This tool does not support task-augmented invocation)rrr#rr r )r r(r'modeerrorr!r!r"validate_task_mode=s zExperimental.validate_task_modetoolcCs |jr|jjnd}|j||dS)a Validate that the request is compatible with the given tool. Convenience wrapper around validate_task_mode that extracts the mode from a Tool. Args: tool: The Tool definition raise_error: If True, raises McpError on validation failure. Returns: None if valid, ErrorData if invalid and raise_error=False Nr&) execution taskSupportr-)r r.r'r+r!r!r"validate_for_toolkszExperimental.validate_for_toolcCs|pt}|tkr |js dSdS)a Check if this client can use a tool with the given task mode. Useful for filtering tool lists or providing warnings. Returns False if tool requires "required" but client doesn't support tasks. Args: tool_task_mode: The tool's execution.taskSupport value Returns: True if the client can use this tool, False otherwise FT)rrr%)r r(r+r!r!r" can_use_tools zExperimental.can_use_tool)task_idmodel_immediate_responseworkr3r4c s|jdur td|jdurtd|jdurtd|j}|j}|j|j|IdH}t||j|j|j|j dd fdd }| |d}|durQt |i}t d d |i|r^d |iSiS) as Create a task, spawn background work, and return CreateTaskResult immediately. This is the recommended way to handle task-augmented tool calls. It: 1. Creates a task in the store 2. Spawns the work function in a background task 3. Returns CreateTaskResult immediately The work function receives a ServerTaskContext with: - elicit() for sending elicitation requests - create_message() for sampling requests - update_status() for progress updates - complete()/fail() for finishing the task When work() returns a Result, the task is auto-completed with that result. If work() raises an exception, the task is auto-failed. Args: work: Async function that does the actual work task_id: Optional task ID (generated if not provided) model_immediate_response: Optional string to include in _meta as io.modelcontextprotocol/model-immediate-response Returns: CreateTaskResult to return to the client Raises: RuntimeError: If task support is not enabled or task_metadata is missing Example: @server.call_tool() async def handle_tool(name: str, args: dict): ctx = server.request_context async def work(task: ServerTaskContext) -> CallToolResult: result = await task.elicit( message="Are you sure?", requestedSchema={"type": "object", ...} ) confirmed = result.content.get("confirm", False) return CallToolResult(content=[TextContent(text="Done" if confirmed else "Cancelled")]) return await ctx.experimental.run_task(work) WARNING: This API is experimental and may change without notice. NzHTask support not enabled. Call server.experimental.enable_tasks() first.zSession not available.zgRequest is not task-augmented (no task field in params). The client must send a task-augmented request.)taskstoresessionqueuehandlerrc szIdH}tjjs|IdHWdSWdStyF}ztjjs;t|IdHWYd}~dSWYd}~dSd}~ww)N)r r6statuscomplete Exceptionfailstr)resultetask_ctxr5r!r"executes  "z&Experimental.run_task..executer6_meta)rNr!) r RuntimeErrorrr task_groupr7 create_taskrr9r: start_soonr r) r r5r3r4supportrGr6rDmetar!rBr"run_tasks2 5   "zExperimental.run_task)__name__ __module__ __qualname____doc__rr__annotations__rrrrr rrpropertyboolr#r%rrr-rr1r2rrrrr?rrLr!r!r!r"r!sL    2 rN) rPcollections.abcrr dataclassesrrtypingr$mcp.server.experimental.task_contextr$mcp.server.experimental.task_supportrmcp.server.sessionr mcp.shared.exceptionsr %mcp.shared.experimental.tasks.helpersr r mcp.typesr rrrrrrrrrrr!r!r!r"s      0