jKhUdZddlmZddlZddlZddlmZddlmZddl m Z m Z ddl m Z ddlZddlmZdd lmZdd lmZdd lmZdd lmZmZmZmZmZdd lmZmZm Z ddl!m"Z#ddl$m%Z%m&Z&m'Z'ddl(m)Z)ddl*m+Z+m,Z,m-Z-ej.e/Z0dZ1de2d< GddZ3GddZ4dS)z/StreamableHTTP Session Manager for MCP servers.) annotationsN)deque) AsyncIterator)AnyFinal)uuid4) TaskStatus)Headers)Request)Response)ASGIAppMessageReceiveScopeSend)AuthenticatedUserAuthorizationContextauthorization_context)Server)MCP_SESSION_ID_HEADER EventStoreStreamableHTTPServerTransport)TransportSecuritySettings)INVALID_REQUEST ErrorData JSONRPCErrori@rDEFAULT_MAX_REQUEST_BODY_SIZEcleZdZdZddddddefd"dZejd#dZd$dZ d$dZ d$d Z d$d!Z dS)%StreamableHTTPSessionManagera^ Manages StreamableHTTP sessions with optional resumability via event store. This class abstracts away the complexity of session management, event storage, and request handling for StreamableHTTP transports. It handles: 1. Session tracking for clients 2. Resumability via an optional event store 3. Connection management and lifecycle 4. Request handling and transport setup 5. Idle session cleanup via optional timeout Important: Only one StreamableHTTPSessionManager instance should be created per application. The instance cannot be reused after its run() context has completed. If you need to restart the manager, create a new instance. Args: app: The MCP server instance event_store: Optional event store for resumability support. If provided, enables resumable connections where clients can reconnect and receive missed events. If None, sessions are still tracked but not resumable. json_response: Whether to use JSON responses instead of SSE streams stateless: If True, creates a completely fresh transport for each request with no session tracking or state persistence between requests. security_settings: Optional transport security settings. retry_interval: Retry interval in milliseconds to suggest to clients in SSE retry field. Used for SSE polling behavior. session_idle_timeout: Optional idle timeout in seconds for stateful sessions. If set, sessions that receive no HTTP requests for this duration will be automatically terminated and removed. When retry_interval is also configured, ensure the idle timeout comfortably exceeds the retry interval to avoid reaping sessions during normal SSE polling gaps. Default is None (no timeout). A value of 1800 (30 minutes) is recommended for most deployments. max_request_body_size: Maximum size in bytes for Streamable HTTP POST request bodies. Requests that exceed this limit receive a 413 response before parsing or session creation. Defaults to 4 MiB. NFappMCPServer[Any, Any] event_storeEventStore | None json_responsebool statelesssecurity_settings TransportSecuritySettings | Noneretry_interval int | Nonesession_idle_timeout float | Nonemax_request_body_sizeintc ||dkrtd|r|td|dkrtd||_||_||_||_||_||_||_||_ t|j ||_ tj|_i|_i|_d|_tj|_d|_dS)Nrz9session_idle_timeout must be a positive number of secondsz7session_idle_timeout is not supported in stateless modez8max_request_body_size must be a positive number of bytesF) ValueError RuntimeErrorr r"r$r&r'r)r+r-RequestBodyLimitMiddleware_handle_requestasgi_appanyioLock_session_creation_lock_server_instances_session_owners _task_group _run_lock _has_started) selfr r"r$r&r'r)r+r-s PC:\Projects\DSCons\.venv\Lib\site-packages\mcp\server\streamable_http_manager.py__init__z%StreamableHTTPSessionManager.__init__Hs +0D0I0IXYY Y  Z-9XYY Y A % %WXX X&*"!2,$8!%:"243GI^__ ',jll#KMAC !returnAsyncIterator[None]c.K|j4d{V|jrtdd|_dddd{Vn#1d{VswxYwYtj4d{V}||_t d dWVt d|j d|_|j |j nq#t d|j d|_|j |j wxYw dddd{VdS#1d{VswxYwYdS)aw Run the session manager with proper lifecycle management. This creates and manages the task group for all session operations. Important: This method can only be called once per instance. The same StreamableHTTPSessionManager instance cannot be reused after this context manager exits. Create a new instance if you need to restart. Use this in the lifespan context manager of your Starlette app: @contextlib.asynccontextmanager async def lifespan(app: Starlette) -> AsyncIterator[None]: async with session_manager.run(): yield NzyStreamableHTTPSessionManager .run() can only be called once per instance. Create a new instance if you need to run again.Tz&StreamableHTTP session manager startedz,StreamableHTTP session manager shutting down) r;r<r1r5create_task_groupr:loggerinfo cancel_scopecancelr8clearr9)r=tgs r>runz StreamableHTTPSessionManager.runqs&> % % % % % % % %  "Y!%D   % % % % % % % % % % % % % % % % % % % % % % % % % % %*,, - - - - - - -!D  KK@ A A A - JKKK&&(((#' &,,...$**,,,,  JKKK&&(((#' &,,...$**,,,,, - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -s=A A  A *"F C?A-F?A.E--F FFscoperreceiversendrNonecDK||||d{VdS)a Process ASGI request with proper session handling and transport setup. Dispatches to the appropriate handler based on stateless mode. Args: scope: ASGI scope receive: ASGI receive function send: ASGI send function N)r4r=rLrMrNs r>handle_requestz+StreamableHTTPSessionManager.handle_requests6 mmE7D11111111111r@cK|jtd|jr||||d{VdS||||d{VdS)Nz6Task group is not initialized. Make sure to use run().)r:r1r&_handle_stateless_request_handle_stateful_requestrQs r>r3z,StreamableHTTPSessionManager._handle_requests   #WXX X > F00FF F F F F F F F F F//wEE E E E E E E E E Er@cdKtdtdjdjt jddfd}jJj|d{V |||d{V d{VdS) z Process request in stateless mode - creating a new transport for each request. Args: scope: ASGI scope receive: ASGI receive function send: ASGI send function z7Stateless mode: Creating new transport for this requestN)mcp_session_idis_json_response_enabledr"r' task_statusrZTaskStatus[None]cK4d{V}|\}}| j||jdd{Vn*#t $rt dYnwxYwdddd{VdS#1d{VswxYwYdS)NTr&zStateless session crashed)connectstartedr rKcreate_initialization_options ExceptionrE exception)rZstreams read_stream write_streamhttp_transportr=s r>run_stateless_serverzTStreamableHTTPSessionManager._handle_stateless_request..run_stateless_servers%--// B B B B B B B7,3) \##%%%B(,,#$>>@@"& ' !BBB$$%@AAAAAB B B B B B B B B B B B B B B B B B B B B B B B B B B B B B Bs4B2;A54B25$BB2BB22 B<?B<)rZr[) rEdebugrr$r'r5TASK_STATUS_IGNOREDr:startrR terminate)r=rLrMrNrgrfs` @r>rTz6StreamableHTTPSessionManager._handle_stateless_requests  NOOO6%)%7"4    KPJc B B B B B B B B B+++$$%9:::::::::++E7DAAAAAAAAA&&(((((((((((r@cL Kt||}|jt}|d}t |t rt |nd}|&|jvrj|}|j|krt d|ddtddttd } t| d d d d } | |||d{VdStd|j,j%t'jjz|j_||||d{VdS|)tdj4d{Vt1j} t5| jjjj jJ||j j< j j<t d| t&j!dd fd} j"Jj"#| d{V |||d{Vdddd{VdS#1d{VswxYwYdStddttd } t| d d d d } | |||d{VdS)z Process request in stateful mode - maintaining session state between requests. Args: scope: ASGI scope receive: ASGI receive function send: ASGI send function userNz\Rejecting request for session %s: credential does not match the one that created the session@z2.0z server-errorzSession not found)codemessage)jsonrpciderrorT)by_alias exclude_noneizapplication/json) status_code media_typez1Session already exists, handling request directlyzCreating new transport)rWrXr"r'r)z'Created new transport with session ID: rYrZr[rArOc.K4d{V}|\}}| tj}j'tjjz|_|_|5j ||j dd{Vdddn #1swxYwY|j rj Jtdj djj djj dd{Vn3#t&$r&tdj dYnwxYwj rej jvrWjsPtdj djj =jj dnt#j rfj jvrYjsStdj djj =jj dwwwwxYwdddd{VdS#1d{VswxYwYdS)NFr]zSession z idle timeoutz crashedzCleaning up crashed session z from active instances.)r^r_r5 CancelScoper+ current_timedeadline idle_scoper rKr`cancelled_caughtrWrErFr8popr9rkrarb is_terminated)rZrcrdrer|rfr=s r> run_serverzIStreamableHTTPSessionManager._handle_stateful_request..run_server0s-5577)^)^)^)^)^)^)^74;1 \#++---&^ */):)<);XZ^ _ _ _ $ 4 8 89VX\ ] ] ]&4&>&>&@&@ @ @ @ @ @ @ @(aaa",,-_8U-_-_-_`````a!/ = ^$2$ATE[$[$[(6(D%\!' %8'5'D%8%8%8!"!"!" %)$:>;X$Y $ 4 8 89VX\ ] ] ]!/ = ^$2$ATE[$[$[(6(D%\!' %8'5'D%8%8%8!"!"!" %)$:>;X$Y $ 4 8 89VX\ ] ] ] ] ^$[$[A)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^)^snJAE<rUz5StreamableHTTPSessionManager._handle_stateful_requests%))!(!4!45J!K!Kyy  3=dDU3V3V`)$///\`  " -2HDLb2b2b./EFID0445KLLLLr*3B3/$!nI?dw#1-1-? $ 0&*&<#'#6 """&4@@@(JSD()FGHV&~'DE VnVVWWWINHa*^*^*^*^*^*^*^*^*^Z'333&,,Z888888888%33E7DIIIIIIIIIGC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JC JL . `s8t8t8tD $$d$FFTWdvH(5'400 0 0 0 0 0 0 0 0 0sCJ11 J;>J;)r r!r"r#r$r%r&r%r'r(r)r*r+r,r-r.)rArBrLrrMrrNrrArO) __name__ __module__ __qualname____doc__rr? contextlibasynccontextmanagerrKrRr3rTrUr@r>rr#s""N*.#>B%)-1%B'"'"'"'"'"R#'-'-'-$#'-R2222$ F F F F/)/)/)/)b~1~1~1~1~1~1r@rc"eZdZdZddZddZdS)r2zIReject oversized HTTP request bodies before invoking an ASGI application.r r max_body_sizer.rArOc"||_||_dSN)r r)r=r rs r>r?z#RequestBodyLimitMiddleware.__init__qs*r@rLrrMrrNrcK|ddks |ddkr|||d{VdSt|}|d}|P t|}||jkr$t dd }|||d{VSn#t $rYnwxYwt}d } d } d} d{V} | dd kr| } nd } | d d} t|t| z|jkr$t dd }|||d{VS| | | dd sd } nt| r' d t|| d|  | dfd }||||d{VdS)NtypehttpmethodPOST)rLzcontent-lengthzRequest body too largei)rvFTz http.requestrr@ more_body)rrrrArcTKrSd{VSr)popleft)cached_messagesrMsr>replayz3RequestBodyLimitMiddleware.__call__..replays9 1&..000 ?????? "r@)rAr) r r rr.rr r0 bytearraylenextendrappendbytes)r=rLrMrNrcontent_length declared_sizer received_bodyreceived_request body_completetrailing_messagerprrrs ` @r>__call__z#RequestBodyLimitMiddleware.__call__us =F " "eHo&?&?((5'400 0 0 0 0 0 0 0 F&&& %566  % @ #N 3 3 !4#555'(@cRRRH!)%$!?!????????6    "   +/ #GIIooooooGv.00#* # ;;vs++D=!!CII-0BBB#$<#NNN%XeWd;;;;;;;;;   & & &;;{E22  $   +0''    " "'}1E1EXeTeff     '  " "#3 4 4 4 # # # # # # # hhufd+++++++++++s$B## B0/B0N)r r rr.rArOr)rrrrr?rrr@r>r2r2nsBSS++++2,2,2,2,2,2,r@r2)5r __future__rrlogging collectionsrcollections.abcrtypingrruuidrr5 anyio.abcr starlette.datastructuresr starlette.requestsr starlette.responsesr starlette.typesr rrrr&mcp.server.auth.middleware.bearer_authrrrmcp.server.lowlevel.serverr MCPServermcp.server.streamable_httprrrmcp.server.transport_securityr mcp.typesrrr getLoggerrrEr__annotations__rr2rr@r>rs555"""""")))))) ,,,,,,&&&&&&((((((BBBBBBBBBBBBBBqqqqqqqqqq:::::: DCCCCC>>>>>>>>>>  8 $ $'66666IH1H1H1H1H1H1H1H1V 9,9,9,9,9,9,9,9,9,9,r@