$j.(UdZddlmZddlZddlZddlZddlZddlmZddl m Z m Z ddl m Z m Z mZddlZe rddlmZddlmZdd lmZejeZd Zd Zd Zd ZdZd'dZd(dZ d)dZ!e"Z#de$d <iZ%d!e$d"<d(d#Z&d*d$Z'd+d&Z(dS),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, session_idstr notificationdict[str, Any]docketr returnNonecK|t|}tj|dt jtj d}| 4d{V}| ||d{V| |td{Vdddd{VdS#1d{VswxYwYdS)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 rr)rattempt enqueued_atN)keyNOTIFICATION_QUEUE_KEYformatjsondumpsrnowrutc isoformatredislpushexpireNOTIFICATION_TTL_SECONDS)rrrrmessager"s j/home/rurouni/.hermes/hermes-agent/venv/lib/python3.11/site-packages/fastmcp/server/tasks/notifications.pypush_notificationr(0s **+22j2II J JCj(#< 55??AA  G||~~:::::::kk#w'''''''''ll3 8999999999::::::::::::::::::::::::::::::s>C"" C,/C,sessionr fastmcpr cK|t|}|t|}td| |4d{V}||dtdzd{Vtt| |gtd{V}|s dddd{V|\}} tj | } | d } | d d } t|| |||d{Vtd || d zn#t $r} | t"d z kri| d z| d <t%| | d<||tj| d{Vtd|| dz| n"td|t"| Yd} ~ nd} ~ wwxYwdddd{Vn#1d{VswxYwYny#t,j$rtd|YdSt $r@}td||t-jd d{VYd}~nd}~wwxYwM)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 %sTN1)ex)timeoutrrrz1Delivered notification to session %s (attempt %d) last_errorz5Requeued notification for session %s (attempt %d): %sz>KrFc K|dd}|dkrtd|tjjd|di|dd}tj|}||d{V|di}|dd kr|di} | d i} | d } | r|d } | st d dSd| vrt d| dS| d} ddl m }tj ||| | | |d| dd}t||tjdSdSdS)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)rKrLrMNstatusinput_requiredz$io.modelcontextprotocol/related-task elicitationtaskIdz:input_required notification missing taskId, skipping relay task_scopez_input_required notification for task %s missing task_scope metadata, skipping elicitation relayr)relay_elicitationzelicitation-relay-name)r9 ValueErrormcptypesTaskStatusNotificationmodel_validateServerNotificationsend_notificationr3r= fastmcp.server.tasks.elicitationrSr> create_task_background_tasksaddadd_done_callbackdiscard)r)rFrrr*rKrserver_notificationrLmeta related_taskrPtask_idrRrStasks r'r:r:sI( " "8-I J JF ---TFTTUUU93BB2'++Hb99&**733  L)66|DD  # #$7 8 88888888 " "8R 0 0F zz(/// $$Wb11xx FKK "&&}55  >jj**G P<//; %l3J J J J J J J&!!':w WUU7'"1"+77D  ! !$ ' ' '  " "#4#< = = = = =50/ > >zset[asyncio.Task[None]]r`z@dict[str, tuple[asyncio.Task[None], weakref.ref[ServerSession]]]_active_subscribersc"K|tvrt|\}}|s |dS|sM|ttj5|d{Vdddn #1swxYwYt|=t jt||||d|dd}|tj |ft|<t d|dS)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-rTrUz.Started notification subscriber for session %s) rjdonecancelrr>r?r_rIweakrefrefr3r4)rr)rr*rh session_refs r'ensure_subscriber_runningrqsU$(((/ ;kyy{{ {{}}8 Fyy{{  KKMMM'011                    +  $Z&'JJ 8 2A2 8 8   D(,W[-A-A&B # LLA:NNNNNs= BBBcRK|tvrdSt|\}}|sM|t t j5|d{Vdddn #1swxYwYtd|dS)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) rjpoprlrmrr>r?r3r4)rrhrDs r'stop_subscriberrts,,,!%%j11GD! 99;; g, - -  JJJJJJJ                LLA:NNNNNs, BBBintc*ttS)z2Get number of active subscribers (for monitoring).)lenrjrir'get_subscriber_country2s " # ##ri)rrrrrr rr) rrr)r rr r*r rr) r)r rFrrrrr r*r rr)rrrr)rru))__doc__ __future__rr>rloggingrn contextlibrrrtypingrrr mcp.typesrXrr mcp.server.sessionr fastmcp.server.serverr getLogger__name__r3rr2r%r<r6r(rIr:r5r`__annotations__rjrqrtryrxrir'rs"#""""" ''''''''++++++++++.000000------  8 $ $>E::::8V#V#V#V#r@>@>@>@>P.1SUU2222  %O%O%O%OPOOOO($$$$$$ri