o c۷i @sdZddlmZmZddlmZmZddlmZmZddl m Z m Z ddl Z ddl mZddlmZmZmZeGd d d ZGd d d eZGd ddeZdS)a) TaskMessageQueue - FIFO queue for task-related messages. This implements the core message queue pattern from the MCP Tasks spec. When a handler needs to send a request (like elicitation) during a task-augmented request, the message is enqueued instead of sent directly. Messages are delivered to the client only through the `tasks/result` endpoint. This pattern enables: 1. Decoupling request handling from message delivery 2. Proper bidirectional communication via the tasks/result stream 3. Automatic status management (working <-> input_required) )ABCabstractmethod) dataclassfield)datetimetimezone)AnyLiteralN)Resolver)JSONRPCNotificationJSONRPCRequest RequestIdc@steZdZUdZeded< eeBed< edddZ e ed< d Z e e eefd Bed < d Zed Bed <d S) QueuedMessagez A message queued for delivery via tasks/result. Messages are stored with their type and a resolver for requests that expect responses. )request notificationtypemessagecCs ttjSN)rnowrutcrra/home/ubuntu/vllm_env/lib/python3.10/site-packages/mcp/shared/experimental/tasks/message_queue.py)s zQueuedMessage.)default_factory timestampNresolveroriginal_request_id)__name__ __module__ __qualname____doc__r __annotations__r r rrrrr dictstrrrr rrrrrs   rc@seZdZdZedededdfddZedededBfdd ZedededBfd d Z edede fd d Z edede efddZ ededdfddZededdfddZdS)TaskMessageQueuea" Abstract interface for task message queuing. This is a FIFO queue that stores messages to be delivered via `tasks/result`. When a task-augmented handler calls elicit() or sends a notification, the message is enqueued here instead of being sent directly to the client. The `tasks/result` handler then dequeues and sends these messages through the transport, with `relatedRequestId` set to the tasks/result request ID so responses are routed correctly. Implementations can use in-memory storage, Redis, etc. task_idrreturnNcdS)z Add a message to the queue for a task. Args: task_id: The task identifier message: The message to enqueue Nr)selfr%rrrrenqueueBzTaskMessageQueue.enqueuecr')z Remove and return the next message from the queue. Args: task_id: The task identifier Returns: The next message, or None if queue is empty Nrr(r%rrrdequeueLr*zTaskMessageQueue.dequeuecr')z Return the next message without removing it. Args: task_id: The task identifier Returns: The next message, or None if queue is empty Nrr+rrrpeekXr*zTaskMessageQueue.peekcr')z Check if the queue is empty for a task. Args: task_id: The task identifier Returns: True if no messages are queued Nrr+rrris_emptydr*zTaskMessageQueue.is_emptycr')a Remove and return all messages from the queue. This is useful for cleanup when a task is cancelled or completed. Args: task_id: The task identifier Returns: All queued messages (may be empty) Nrr+rrrclearpr*zTaskMessageQueue.clearcr')z Wait until a message is available in the queue. This blocks until either: 1. A message is enqueued for this task 2. The wait is cancelled Args: task_id: The task identifier Nrr+rrrwait_for_message~r*z!TaskMessageQueue.wait_for_messagecr')z Signal that a message is available for a task. This wakes up any coroutines waiting in wait_for_message(). Args: task_id: The task identifier Nrr+rrrnotify_message_availabler*z)TaskMessageQueue.notify_message_available)rrrr rr#rr)r,r-boolr.listr/r0r1rrrrr$3s       r$c@seZdZdZdddZdedeefddZded eddfd d Z dededBfd d Z dededBfddZ dede fddZ dedeefddZdeddfddZdeddfddZddedBddfddZdS)InMemoryTaskMessageQueueaA In-memory implementation of TaskMessageQueue. This is suitable for single-process servers. For distributed systems, implement TaskMessageQueue with Redis, RabbitMQ, etc. Features: - FIFO ordering per task - Async wait for message availability - Thread-safe for single-process async use r&NcCsi|_i|_dSr)_queues_events)r(rrr__init__s z!InMemoryTaskMessageQueue.__init__r%cCs||jvr g|j|<|j|S)z#Get or create the queue for a task.)r5r+rrr _get_queues   z#InMemoryTaskMessageQueue._get_queuercs*||}||||IdHdS)zAdd a message to the queue.N)r8appendr1)r(r%rqueuerrrr)s  z InMemoryTaskMessageQueue.enqueuecs||}|s dS|dS)z#Remove and return the next message.Nr)r8popr(r%r:rrrr,s   z InMemoryTaskMessageQueue.dequeuecs||}|s dS|dS)z,Return the next message without removing it.Nr)r8r<rrrr-s  zInMemoryTaskMessageQueue.peekcs||}t|dkS)zCheck if the queue is empty.r)r8lenr<rrrr.s  z!InMemoryTaskMessageQueue.is_emptycs ||}t|}||S)zRemove and return all messages.)r8r3r/)r(r%r:messagesrrrr/s  zInMemoryTaskMessageQueue.clearcsT||IdHs dSt|j|<|j|}||IdHs!dS|IdHdS)z"Wait until a message is available.N)r.anyioEventr6wait)r(r%eventrrrr0s z)InMemoryTaskMessageQueue.wait_for_messagecs"||jvr|j|dSdS)z#Signal that a message is available.N)r6setr+rrrr1s z1InMemoryTaskMessageQueue.notify_message_availablecCs@|dur|j|d|j|ddS|j|jdS)z Clean up queues and events. Args: task_id: If provided, clean up only this task. Otherwise clean up all. N)r5r;r6r/r+rrrcleanups  z InMemoryTaskMessageQueue.cleanup)r&Nr)rrrr r7r#r3rr8r)r,r-r2r.r/r0r1rDrrrrr4s r4)r abcrr dataclassesrrrrtypingrr r?&mcp.shared.experimental.tasks.resolverr mcp.typesr r r rr$r4rrrrs d