jdZddlmZmZddlmZmZddlmZmZddl m Z m Z ddl Z ddl mZddlmZmZmZeGd d ZGd d eZGd deZdS)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 RequestIdceZdZUdZeded< eezed< edZ e ed< dZ e e eefdzed < dZedzed <dS) QueuedMessagez A message queued for delivery via tasks/result. Messages are stored with their type and a resolver for requests that expect responses. )request notificationtypemessagec>tjtjSN)rnowrutcYC:\Projects\DSCons\.venv\Lib\site-packages\mcp\shared\experimental\tasks\message_queue.pyzQueuedMessage.)s X\8R8Rr)default_factory timestampNresolveroriginal_request_id)__name__ __module__ __qualname____doc__r __annotations__r r rrrrr dictstrr r rrrrrrs + ,,,,Q 1 1111'%0R0RSSSIxSSS(04HhtCH~&-444D,0T)000NNrrceZdZdZedededdfdZedededzfdZedededzfdZ edede fd Z edede efd Z ededdfd Zededdfd ZdS) 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_idrreturnNc KdS)z Add a message to the queue for a task. Args: task_id: The task identifier message: The message to enqueue Nr)selfr*rs renqueuezTaskMessageQueue.enqueueB rc KdS)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*s rdequeuezTaskMessageQueue.dequeueLr/rc KdS)z Return the next message without removing it. Args: task_id: The task identifier Returns: The next message, or None if queue is empty Nrr1s rpeekzTaskMessageQueue.peekXr/rc KdS)z Check if the queue is empty for a task. Args: task_id: The task identifier Returns: True if no messages are queued Nrr1s ris_emptyzTaskMessageQueue.is_emptydr/rc KdS)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) Nrr1s rclearzTaskMessageQueue.clearpr/rc KdS)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 Nrr1s rwait_for_messagez!TaskMessageQueue.wait_for_message~r/rc KdS)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 Nrr1s rnotify_message_availablez)TaskMessageQueue.notify_message_availabler/r)r!r"r#r$rr'rr.r2r4boolr6listr8r:r<rrrr)r)3sv   S = T   ^   S  ]T-A    ^    #  -$*>    ^    c  d    ^    3  4 +>    ^    c  d    ^   c d   ^   rr)ceZdZdZddZdedeefdZdededdfdZ dededzfd Z dededzfd Z dede fd Z dedeefd Zdeddfd ZdeddfdZddedzddfdZdS)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+Nc"i|_i|_dSr)_queues_events)r-s r__init__z!InMemoryTaskMessageQueue.__init__s79 /1 rr*cB||jvr g|j|<|j|S)z#Get or create the queue for a task.)rBr1s r _get_queuez#InMemoryTaskMessageQueue._get_queues( $, & &$&DL !|G$$rrcK||}||||d{VdS)zAdd a message to the queue.N)rFappendr<)r-r*rqueues rr.z InMemoryTaskMessageQueue.enqueuesU(( W++G44444444444rcbK||}|sdS|dS)z#Remove and return the next message.Nr)rFpopr-r*rIs rr2z InMemoryTaskMessageQueue.dequeues4(( 4yy||rcHK||}|sdS|dS)z,Return the next message without removing it.Nr)rFrLs rr4zInMemoryTaskMessageQueue.peeks-(( 4QxrcVK||}t|dkS)zCheck if the queue is empty.r)rFlenrLs rr6z!InMemoryTaskMessageQueue.is_emptys'((5zzQrczK||}t|}||S)zRemove and return all messages.)rFr>r8)r-r*rImessagess rr8zInMemoryTaskMessageQueue.clears4((;; rcK||d{VsdStj|j|<|j|}||d{VsdS|d{VdS)z"Wait until a message is available.N)r6anyioEventrCwait)r-r*events rr:z)InMemoryTaskMessageQueue.wait_for_messages]]7++++++++  F!&  W W%]]7++++++++  Fjjllrc^K||jvr!|j|dSdS)z#Signal that a message is available.N)rCsetr1s rr<z1InMemoryTaskMessageQueue.notify_message_availables: dl " " L ! % % ' ' ' ' ' # "rc|8|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)rBrKrCr8r1s rcleanupz InMemoryTaskMessageQueue.cleanupsn   L  Wd + + + L  Wd + + + + + L   L   r)r+Nr)r!r"r#r$rDr'r>rrFr.r2r4r=r6r8r:r<rZrrrr@r@s  2222%#%$}*=%%%% 5S5=5T5555S]T-A#-$*>cd 34 +>cd"(c(d(((( ! !sTz !T ! ! ! ! ! !rr@)r$abcrr dataclassesrrrrtypingr r rS&mcp.shared.experimental.tasks.resolverr mcp.typesr r rrr)r@rrrr`sX  $#######(((((((('''''''' ;;;;;;DDDDDDDDDD OOOOOOO O0a a a a a sa a a HZ!Z!Z!Z!Z!/Z!Z!Z!Z!Z!r