i,tUdZddlmZddlZddlZddlZddlZddlmZddl m Z m Z ddl m Z m Z mZddlZe rddlmZddlmZdd lmZej.eZd Zd Zd Zd ZdZ ddZ ddZ ddZ!e"Z#de$d<iZ%de$d< ddZ&ddZ'ddZ(y)aDistributed notification queue for background task events (SEP-1686). Enables distributed Docket workers to send MCP notifications to clients without holding session references. Workers push to a Redis queue, the MCP server process subscribes and forwards to the client's session. Pattern: Fire-and-forward with retry - One queue per session_id - LPUSH/BRPOP for reliable ordered delivery - Retry up to 3 times on delivery failure, then discard - TTL-based expiration for stale messages Note: Docket's execution.subscribe() handles task state/progress events via Redis Pub/Sub. This module handles elicitation-specific notifications that require reliable delivery (input_required prompts, cancel signals). ) annotationsN)suppress)datetimetimezone) TYPE_CHECKINGAnycast)Docket) ServerSession)FastMCPz"fastmcp:notifications:{session_id}z)fastmcp:notifications:{session_id}:activei,cK|jtj|}tj|dt j tjjd}|j4d{}|j||d{|j|td{dddd{y7M7577 #1d{7swYyxYww)a~Push notification to session's queue (called from Docket worker). Used for elicitation-specific notifications (input_required, cancel) that need reliable delivery across distributed processes. Args: session_id: Target session's identifier notification: MCP notification dict (method, params, _meta) docket: Docket instance for Redis access  session_idr) notificationattempt enqueued_atN)keyNOTIFICATION_QUEUE_KEYformatjsondumpsrnowrutc isoformatredislpushexpireNOTIFICATION_TTL_SECONDS)rrdocketrmessagers r/Users/bowang/.openclaw/workspace/ChatDev/.venv/lib/python3.12/site-packages/fastmcp/server/tasks/notifications.pypush_notificationr$0s **+22j2I JCjj(#<< 5??A G||~~kk#w'''ll3 8999~~'9~~~slBC/CC/CCC<C=C C/ C C/CCC/C, C# !C,(C/cK|jtj|}|jtj|}tj d| |j 4d{}|j|dtdzd{tt|j|gtd{}|s dddd{|\}} tj| } | d} | jd d } t|| |||d{tj d || d zdddd{777777#t $r} | t"d z kr`| d z| d <t%| | d <|j'|tj(| d{7tj d|| dz| ntj+d|t"| Yd} ~ d} ~ wwxYw7#1d{7swYxYw#t,j.$rtj d|Yyt $r@}tj d||t-j0d d{7Yd}~.d}~wwxYww)a~Subscribe to notification queue and forward to session. Runs in the MCP server process. Bridges distributed workers to clients. This loop: 1. Maintains a heartbeat (active subscriber marker for debugging) 2. Blocks on BRPOP waiting for notifications 3. Forwards notifications to the client's session 4. Retries failed deliveries, then discards (no dead-letter queue) Args: session_id: Session identifier to subscribe to session: MCP ServerSession for sending notifications docket: Docket instance for Redis access fastmcp: FastMCP server instance (for elicitation relay) rz/Starting notification subscriber for session %sN1)ex)timeoutrrrz1Delivered notification to session %s (attempt %d) last_errorz5Requeued notification for session %s (attempt %d): %szJ J  J c 4K|jdd}|dk7rtd|tjjj d|jdi|jdd}tjj |}|j|d{|jdi}|jdd k(r|jdi} | jd i} | jd } | r|jd } | stjd yddl m } tj| ||| | |d| dd}tj||j!tj"yyy7w)a0Reconstruct MCP notification from dict and send to session. For input_required notifications with elicitation metadata, also sends a standard elicitation/create request to the client and relays the response back to the worker via Redis. Args: session: MCP ServerSession notification_dict: Notification as dict (method, params, _meta) session_id: Session identifier (for elicitation relay) docket: Docket instance (for notification delivery) fastmcp: FastMCP server instance (for elicitation relay) methodznotifications/tasks/statusz0Unsupported notification method for subscriber: params_meta)rHrIrJNstatusinput_requiredz$io.modelcontextprotocol/related-task elicitationtaskIdz:input_required notification missing taskId, skipping relayr)relay_elicitationzelicitation-relay-name)r3 ValueErrormcptypesTaskStatusNotificationmodel_validateServerNotificationsend_notificationr-r8 fastmcp.server.tasks.elicitationrOr9 create_task_background_tasksaddadd_done_callbackdiscard)r<rCrr!r=rHrserver_notificationrImeta related_taskrMtask_idrOtasks r#r4r4s( " "8-I JF --KF8TUU9933BB2'++Hb9&**73 L))66|D  # #$7 888 " "8R 0F zz(// $$Wb1xx FK "&&}5 jj*GP J&&!':w WU)'"1+7D  ! !$ '  " "#4#<#< =  0 9sB%F'F(C/Fzset[asyncio.Task[None]]r\z@dict[str, tuple[asyncio.Task[None], weakref.ref[ServerSession]]]_active_subscriberscK|tvrwt|\}}|js|y|js<|jttj 5|d{dddt|=t j t||||d|dd}|tj|ft|<tjd|y7q#1swYpxYww)aStart notification subscriber if not already running (idempotent). Subscriber is created on first task submission and cleaned up on disconnect. Safe to call multiple times for the same session. Args: session_id: Session identifier session: MCP ServerSession docket: Docket instance fastmcp: FastMCP server instance (for elicitation relay) Nznotification-subscriber-rPrQz.Started notification subscriber for session %s) redonecancelrr9r:r[rFweakrefrefr-r.)rr<r!r=rd session_refs r#ensure_subscriber_runningrls$((/ ;kyy{{}8 yy{ KKM'001 2  +   $Z&'J' 2A'7 8 D(,W[[-A&B # LLA:N21s1A&C.(C".C /C"3A-C. C""C+'C.c.K|tvrytj|\}}|js<|jt t j 5|d{dddtjd|y7##1swY"xYww)zStop notification subscriber for a session. Called when session disconnects. Pending messages remain in queue for delivery if client reconnects (with TTL expiration). Args: session_id: Session identifier Nz.Stopped notification subscriber for session %s) repoprgrhrr9r:r-r.)rrdrAs r#stop_subscriberrosq,,!%%j1GD! 99; g,, -JJ. LLA:N . -s0ABB #B$B (BB  BBc ttS)z2Get number of active subscribers (for monitoring).)lenrer#get_subscriber_countrt*s " ##rs)rr7rdict[str, Any]r!r returnNone) rr7r<r r!r r=r rvrw) r<r rCrurr7r!r r=r rvrw)rr7rvrw)rvint))__doc__ __future__rr9rloggingri contextlibrrrtypingrrr mcp.typesrTr!r mcp.server.sessionr fastmcp.server.serverr getLogger__name__r-rr,r r6r0r$rFr4r/r\__annotations__rerlrortrrrsr#rst"# '++0-   8 $>E:: : : :8V#V# V# V# V#  V#r8> 8>%8>8>  8>  8>  8>@.1U*2  %O%O %O %O %O  %OPO($rs