o i/@sdZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl Z ddlmZdd lmZdd lmZdd lmZmZmZdd lmZdd lmZmZmZddlm Z ddl!m"Z"m#Z#m$Z$e%e&Z'GdddZ(dS)z/StreamableHTTP Session Manager for MCP servers.) annotationsN) AsyncIterator) HTTPStatus)Any)uuid4) TaskStatus)Request)Response)ReceiveScopeSend)Server)MCP_SESSION_ID_HEADER EventStoreStreamableHTTPServerTransport)TransportSecuritySettings)INVALID_REQUEST ErrorData JSONRPCErrorc@sTeZdZdZ     d"d#ddZejd$ddZd%ddZd%ddZ d%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 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. NFappMCPServer[Any, Any] event_storeEventStore | None json_responsebool statelesssecurity_settings TransportSecuritySettings | Noneretry_interval int | NonecCsN||_||_||_||_||_||_t|_i|_ d|_ t|_ d|_ dS)NF) rrrrrranyioLock_session_creation_lock_server_instances _task_group _run_lock _has_started)selfrrrrrrr)V/home/ubuntu/.local/lib/python3.10/site-packages/mcp/server/streamable_http_manager.py__init__<s    z%StreamableHTTPSessionManager.__init__returnAsyncIterator[None]c Cs|j4IdH|jrtdd|_WdIdHn 1IdHs#wYt4IdH?}||_tdzdVWtd|j d|_|j ntd|j d|_|j wWdIdHdS1IdHsuwYdS)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' RuntimeErrorr!create_task_groupr%loggerinfo cancel_scopecancelr$clear)r(tgr)r)r*runVs,(     .z StreamableHTTPSessionManager.runscoper receiver sendr NonecsJ|jdur td|jr||||IdHdS||||IdHdS)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 Nz6Task group is not initialized. Make sure to use run().)r%r.r_handle_stateless_request_handle_stateful_request)r(r7r8r9r)r)r*handle_requests  z+StreamableHTTPSessionManager.handle_requestcs~tdtdjdjdtjdd fdd}jdus#Jj|IdH |||IdH IdHdS) 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_enabledrr task_statusrATaskStatus[None]c s4IdH@}|\}}|zjj||jddIdHWnty2tdYn wWdIdHdSWdIdHdS1IdHsOwYdS)NTrzStateless session crashed)connectstartedrr6create_initialization_options Exceptionr0 exception)rAstreams read_stream write_streamhttp_transportr(r)r*run_stateless_servers$  .zTStreamableHTTPSessionManager._handle_stateless_request..run_stateless_server)rArB) r0debugrrrr!TASK_STATUS_IGNOREDr%startr= terminate)r(r7r8r9rNr)rLr*r;s z6StreamableHTTPSessionManager._handle_stateless_requestc st||}|jt}|dur+|jvr+j|}td||||IdHdS|durtdj4IdHXt j }t |j j jjdjdusTJjj<td|tjddfd d }jdusuJj|IdH|||IdHWdIdHdS1IdHswYdStd dttddd} t| jdddtjdd} | |||IdHdS)z Process request in stateful mode - maintaining session state between requests. Args: scope: ASGI scope receive: ASGI receive function send: ASGI send function Nz1Session already exists, handling request directlyzCreating new transport)r>r?rrrz'Created new transport with session ID: r@rArBr,r:c s.4IdH}|\}}|zPzjj||jddIdHWn tyE}ztjdjd|ddWYd}~nd}~wwWjrbjj vrbj sbt djdj j=njr~jj vr~j s~t djdj j=wWdIdHdS1IdHswYdS) NFrCzSession z crashed: T)exc_infozCleaning up crashed session z from active instances.) rDrErr6rFrGr0errorr>r$ is_terminatedr1)rArIrJrKerLr)r* run_serversV   .zIStreamableHTTPSessionManager._handle_stateful_request..run_serverz2.0z server-errorzSession not found)codemessage)jsonrpcidrTT)by_alias exclude_nonezapplication/json)content status_code media_type)rArBr,r:)rheadersgetrr$r0rOr=r#rhexrrrrrr>r1r!rPr%rQrrrr model_dump_jsonr NOT_FOUND) r(r7r8r9requestrequest_mcp_session_id transportnew_session_idrWerror_responseresponser)rLr*r<sR      .8 z5StreamableHTTPSessionManager._handle_stateful_request)NFFNN) rrrrrrrrrrrr )r,r-)r7r r8r r9r r,r:) __name__ __module__ __qualname____doc__r+ contextlibasynccontextmanagerr6r=r;r<r)r)r)r*rs    ( 1r))ro __future__rrploggingcollections.abcrhttprtypingruuidrr! anyio.abcrstarlette.requestsrstarlette.responsesr starlette.typesr r r mcp.server.lowlevel.serverr MCPServermcp.server.streamable_httprrrmcp.server.transport_securityr mcp.typesrrr getLoggerrlr0rr)r)r)r*s&