xj#+.dZddlmZddlZddlmZddlmZmZddl Z ddl Z ddl Z ddl m Z mZmZmZmZmZmZddlmZddlmZdd lmZmZmZmZmZmZmZm Z gd Z!ejDe#Z$Gd d eZ%Gd dZ&ddZ'ddZ(GddZ)y)a&Server-side `subscriptions/listen` support (2026-07-28, SEP-2575). On the 2026-07-28 wire there is no standing GET stream: a client opts in to server events by sending a `subscriptions/listen` request whose response IS the stream. This module provides the two pieces a server needs: - `SubscriptionBus`: the pluggable fan-out seam. The bus carries typed `ServerEvent` values, not wire notifications - the listen handler owns subscription-id stamping and per-stream filtering, so a custom bus (e.g. backed by Redis pub/sub for multi-replica deployments) never sees JSON-RPC. The in-process default is `InMemorySubscriptionBus`. - `ListenHandler`: the request handler that serves `subscriptions/listen`. `MCPServer` registers one automatically; lowlevel `Server` users pass an instance as `on_subscriptions_listen=`. The event vocabulary lives in `mcp.shared.subscriptions`, shared with the client driver, and is re-exported here. Per the spec, the handler acknowledges first (the ack is the first frame on the stream), tags every frame with the listen request's JSON-RPC id under `_meta["io.modelcontextprotocol/subscriptionId"]`, and never delivers an event kind the client did not request. Delivery is fire-and-forget with no replay: a dropped stream is not resumable - clients re-listen and refetch. ) annotationsN)Callable)AnyProtocol)INTERNAL_ERRORINVALID_REQUESTSubscriptionFilter%SubscriptionsAcknowledgedNotification+SubscriptionsAcknowledgedNotificationParams SubscriptionsListenRequestParamsSubscriptionsListenResult)ServerRequestContext)MCPError)SUBSCRIPTION_ID_META_KEYPromptsListChangedResourcesListChangedResourceUpdated ServerEventToolsListChanged event_matchesevent_to_notification) rInMemorySubscriptionBus ListenHandlerrrrrSubscriptionBusrc eZdZdZddZddZy)raEFan-out seam between event publishers and open listen streams. Implement this over an external pub/sub backend (Redis, NATS, ...) to fan events out across replicas: `publish` forwards the event to the backend, and each replica's bus invokes its local listeners for events arriving from the backend. The same instance can be shared across servers. `publish` is async so backend implementations can do network I/O. `subscribe` is synchronous local registration. Listeners are synchronous, must not raise, and are invoked on the server's event loop. c Kyw)z-Deliver `event` to every subscribed listener.N)selfevents l/mnt/ssd/data/Dropbox/adrian/sandbox/mcp-query/venv/lib/python3.12/site-packages/mcp/server/subscriptions.pypublishzSubscriptionBus.publishUs  scy)BRegister `listener` and return an idempotent unsubscribe callable.Nr)rlisteners r subscribezSubscriptionBus.subscribeYs NrrreturnNoner$zCallable[[ServerEvent], None]r(Callable[[], None])__name__ __module__ __qualname____doc__r!r%rr&r rrHs   r&rc(eZdZdZddZddZddZy) rzUIn-process `SubscriptionBus`: synchronous fan-out to listeners in subscription order.ci|_yN) _listeners)rs r __init__z InMemorySubscriptionBus.__init__as HJr&cKt|jjD] } || t jjd{y#t$rtj dYUwxYw7)w)a`Deliver `event` to every subscribed listener. A raising listener is logged and skipped: one bad listener must not starve the others or fail the publishing handler. Ends with a checkpoint so a burst of publishes from one task lets listen streams drain between events instead of overflowing their buffers unread. z(subscription listener raised; continuingN) listr3values Exceptionlogger exceptionanyiolowlevel checkpoint)rrr$s r r!zInMemorySubscriptionBus.publishfsuT__3356 MH M M nn''))) M  !KL M)s3'BA#BA?BA<9B;A<<BcJt|j<dfd }|S)r#c>jjdyr2)r3pop)rtokensr unsubscribez6InMemorySubscriptionBus.subscribe..unsubscribezs OO  t ,r&r(r))objectr3)rr$rBrAs` @r r%z!InMemorySubscriptionBus.subscribeus%!) -r&NrCr'r*)r,r-r.r/r4r!r%rr&r rr^s_J *r&rc\ |y#t$rtjdYywxYw)aRun a bus's unsubscribe callable, isolating the stream from it raising. The callable comes from a custom `SubscriptionBus`; a raising one is logged and skipped so it cannot stop the stream's own cleanup from releasing its subscription slot. z1bus unsubscribe raised; continuing stream cleanupN)r8r9r:)rBs r _safe_unsubscriberFs-N NLMNs  ++ct|jrdnd|jrdnd|jrdnd|jrt |jSdS)agThe subset of `requested` the server will deliver, for the ack. Every requested kind is honored - whether an event kind ever fires depends on what the server publishes, exactly as a subscription to a nonexistent resource URI is honored and never fires. Non-true flags and an empty URI list are dropped rather than echoed as falsy values. TN)tools_list_changedprompts_list_changedresources_list_changedresource_subscriptions)r rHrIrJrKr6) requesteds r _honored_subsetrMs_ #,#?#?4T%.%C%CT'0'G'GtTIRIiIitI$D$DE  pt  r&c<eZdZdZdddddZ d dZd dZy) raServes `subscriptions/listen`: one call is one subscription stream. Register on a lowlevel `Server` via `on_subscriptions_listen=` (or `add_request_handler`); `MCPServer` does so automatically. Each call acknowledges the honored filter first, then forwards matching bus events onto the request's response stream until the client disconnects (which cancels the handler; the stream just ends, per the spec's abrupt-close contract) or `close` ends all streams gracefully. Served on any transport that can carry the request's response stream: streamable HTTP's SSE mode, or a duplex stream pair such as stdio. `max_subscriptions` bounds concurrent streams (further listen requests are rejected with `INTERNAL_ERROR`, before the ack). `max_buffered_events` bounds each stream's event backlog: a stream whose client has stopped reading is ended at the cap (the client re-listens and refetches - there is no replay, so ending the stream loses nothing the backlog wasn't already losing). i)max_subscriptionsmax_buffered_eventscL||_||_||_t|_yr2)_bus_max_subscriptions_max_buffered_eventsset_streams)rbusrOrPs r r4zListenHandler.__init__s# "3$7!WZW\ r&c* K|j ttdtjj k\rtt dt|jtjxsd t i}tjtj\ }d  fd }j j#|}jj%  |j&j)t+t-| d{|23d{}|j&j)t/|| d{97?767 6 t1|jj3  j5|j5nK#t1|jj3  j5|j5wxYwt7| Sw) zServe one listen stream.Nz*subscriptions/listen requires a request idzSubscription limit reachedrc"t|r j|yy#tj$rYytj$rDt j djjjYywxYw)Nz0listen stream %r backlog full; ending the stream) r send_nowaitr;ClosedResourceError WouldBlockr9warningrVdiscardclose)rhonored honored_urisrsendsubscription_ids r deliverz'ListenHandler.__call__..delivers|WlE: !$$U+;00''!NN#UWfgMM))$/JJL !s#BAB B) notifications_meta)params)related_request_id)rfr') request_idrrlenrVrSrrMre frozensetrKrr;create_memory_object_streamrrTrRr%addsessionsend_notificationr r rrFr^r_r ) rctxrgmetarecvrdrBrr`rarbrcs ` @@@@r __call__zListenHandler.__call__s ..  "?,XY Y t}} !8!8 8>+GH H!&"6"67 !?!?!E2F 8/J 66{CDD]D]^ d ! !&ii))'2  $ ++//5FU\dhi$3 0   $  ekk33)%6?4    $ k * MM ! !$ ' JJL JJL k * MM ! !$ ' JJL JJL(t44sbC0H85F<-E-.F<4E38E/9E3<+F<'E1(F</E31F<3F<5AH(JNjn] 95 +95195 # 95v r&r)rBr+r(r))rLr r(r )*r/ __future__rloggingcollections.abcrtypingrrr;anyio.lowlevelanyio.streams.memory mcp_typesrrr r r r r mcp.server.contextrmcp.shared.exceptionsrmcp.shared.subscriptionsrrrrrrrr__all__ getLoggerr,r9rrrFrMrrr&r rs0#$ 4*       8 $ h ,D N  aar&