+ Q(i,Pa0t$Rt^RIHt^RIt^RIt^RIt^RIt^RIH t ^RI H t H t ^RI H t HtHt^RIt] 'd^RIHt^RIHt^RIHt]P0!]4tR tR tR t^t^tR R lt RRlt!RRlt"]#!4t$R]%R&/t&R]%R&RRlt'RRlt(RRlt)R#)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,c(V^8dQhRRRRRRRR/#)  session_idstr notificationdict[str, Any]docketr returnNone)formats"h/Users/agent/.openclaw/workspace/venv/lib/python3.14/site-packages/fastmcp/server/tasks/notifications.py __annotate__r0s0::: : : :c  "VP\PVR74p\P!RVR^R\ P !\P4P4/4pVP4;_uu_4GRjxL pVPW44GRjxL VPV\4GRjxL RRR4GRjxL R#LML6LL +GRjxL 'giR#;i5i)abPush 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 rrattempt enqueued_atN)keyNOTIFICATION_QUEUE_KEYrjsondumpsrnowrutc isoformatredislpushexpireNOTIFICATION_TTL_SECONDS)rrrrmessager%s&&& rpush_notificationr*0s **+22j2I JCjj L q 8<< 5??A G||~~~kk#'''ll3 8999~~'9~~~srBDCDC&)C *C&C"C& DC$D C&"C&$D&D ,C/- D 8D : Dc ,V^8dQhRRRRRRRRR R /# r rrsessionr rr fastmcpr rrr)rs"rrrLsAV#V#V# V# V# V#  V#rc "VP\PVR74pVP\PVR74p\P RV4VP 4;_uu_4GRjxL pVPVR\^,R7GRjxL \\VPV.\R74GRjxL pV'gRRR4GRjxL KVwr\P!V 4p V R,p V PR^4p \WWV4GRjxL \P R VV ^,4RRR4GRjxL EKLLLLL= \ dp T \"^, 8dnT ^,T R&\%T 4T R &TP'T\P(!T 44GRjxL \P R TT ^,T 4Rp ? L\P+R T\"T 4Rp ? LRp ? ii;iL +GRjxL 'giEK;i \,P.d\P R T4R#\ dBp\P RY4\,P0!^4GRjxL Rp?EK`Rp?ii;i5i)aNSubscribe 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)timeoutrrz1Delivered notification to session %s (attempt %d) last_errorz5Requeued notification for session %s (attempt %d): %sz8> 8>%8>8>  8>  8>  8>rc h"VPRR4pVR8wd\RV 24h\PPP RRRVPR/4RVPR4/4p\PP V4pVPV4GRjxL VPR/4pVPR4R8XdVPR/4p V PR /4p V PR 4p V 'dVPR 4p V 'g\PR 4R#^R I H p \P!V !WWV4RV R, 2R7p\PV4VP!\P"4R#R#R#EL5i)aReconstruct 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_metaNstatusinput_requiredz$io.modelcontextprotocol/related-task elicitationtaskIdz:input_required notification missing taskId, skipping relay)relay_elicitationzelicitation-relay-NNname)r; ValueErrormcptypesTaskStatusNotificationmodel_validateServerNotificationsend_notificationr5r? fastmcp.server.tasks.elicitationrUr@ create_task_background_tasksaddadd_done_callbackdiscard)r-rHrrr.rNrserver_notificationrOmeta related_taskrStask_idrUtasks&&&&& rr<r<s( " "8-I JF --KF8TUU9933BB 2 '++Hb9 &**73 L))66|D  # #$7 888 " "8R 0F zz(// $$Wb1xx FK "&&}5 jj*GP J&&!'wWU)'"+7D  ! !$ '  " "#4#<#< =  0 9s B)F2+F/,A(F2F2.BF2zset[asyncio.Task[None]]rcz@dict[str, tuple[asyncio.Task[None], weakref.ref[ServerSession]]]_active_subscribersc ,V^8dQhRRRRRRRRR R /#r,r)rs"rrrsA%O%O%O %O %O %O  %Orc2"V\9d\V,wrEVP4'g V!4eR#VP4'gEVP4\\P 4;_uu_4VGRjxL RRR4\V\P !\WW#4RVR, 2R7pV\P!V43\V&\PRV4R#Lv +'giL|;i5i)aeStart 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-rVrXz.Started notification subscriber for session %s) rldonecancelrr@rArbrKweakrefrefr5r6)rr-rr.rk session_refs&&&& rensure_subscriber_runningrts$((/ ;yy{{{}8 yy{{ KKM'0011 2  +   $Z&J' 2'7 8 D(,W[[-A&B # LLA:N21s7AD1DD D DA2DD D Dc V^8dQhRRRR/#)r rrrrr)rs"rrrsOOcOdOrc`"V\9dR#\PV4wrVP4'gEVP4\ \ P 4;_uu_4VGRjxL RRR4\PRV4R#L$ +'giL*;i5i)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) rlpoprorprr@rAr5r6)rrkrFs& rstop_subscriberrxsr,,!%%j1GD 99;; g,, - -JJ. LLA:N . -s0A,B..B4B5B9 B.B B+ &B.cV^8dQhRR/#)r rintr)rs"rrr*s$$c$rc \\4#)z2Get number of active subscribers (for monitoring).)lenrlrrrget_subscriber_countr}*s " ##r)*__conditional_annotations____doc__ __future__rr@r loggingrq contextlibrrrtypingrrr mcp.typesr[rr mcp.server.sessionr fastmcp.server.serverr getLogger__name__r5rr4r(r>r8r*rKr<r7rc__annotations__rlrtrxr})r~s@rrs"# '++0-   8 $>E:8V#r8>@.1U*2  %OPO($r