o h+@sdZddlmZddlZddlZddlmZddlmZddl m Z ddl m Z ddl Z ddlmZdd lmZdd lmZdd lmZmZmZdd lmZdd lmZmZmZddlm 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)TransportSecuritySettingsc@sReZdZdZ    d d!d dZejd"ddZd#ddZd#ddZ d#ddZ 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. NFappMCPServer[Any, Any] event_storeEventStore | None json_responsebool statelesssecurity_settings TransportSecuritySettings | NonecCsH||_||_||_||_||_t|_i|_d|_ t|_ d|_ dS)NF) rrrrranyioLock_session_creation_lock_server_instances _task_group _run_lock _has_started)selfrrrrrr$c/var/www/html/openai_agents/venv/lib/python3.10/site-packages/mcp/server/streamable_http_manager.py__init__8s   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" RuntimeErrorrcreate_task_groupr loggerinfo cancel_scopecancelrclear)r#tgr$r$r%runPs,(     .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#r2r3r4r$r$r%handle_requestys  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 requestNmcp_session_idis_json_response_enabledrr task_statusr=TaskStatus[None]c s4IdH@}|\}}|zjj||jddIdHWnty2tdYn wWdIdHdSWdIdHdS1IdHsOwYdS)NTrzStateless session crashed)connectstartedrr1create_initialization_options Exceptionr+ exception)r=streams read_stream write_streamhttp_transportr#r$r%run_stateless_servers$  .zTStreamableHTTPSessionManager._handle_stateless_request..run_stateless_server)r=r>) r+debugrrrrTASK_STATUS_IGNOREDr startr8 terminate)r#r2r3r4rJr$rHr%r6s z6StreamableHTTPSessionManager._handle_stateless_requestc s`t||}|jt}|dur+|jvr+j|}td||||IdHdS|durtdj4IdHVt j }t |j j jdjdusRJjj<td|tjddfd d }jdussJj|IdH|||IdHWdIdHdS1IdHswYdStd tjd} | |||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 transportr9z'Created new transport with session ID: r<r=r>r'r5c s.4IdH}|\}}|zPzjj||jddIdHWn tyE}ztjdjd|ddWYd}~nd}~wwWjrbjj vrbj sbt djdj j=njr~jj vr~j s~t djdj j=wWdIdHdS1IdHswYdS) NFr?zSession z crashed: T)exc_infozCleaning up crashed session z from active instances.) r@rArr1rBrCr+errorr:r is_terminatedr,)r=rErFrGerHr$r% run_serversV   .zIStreamableHTTPSessionManager._handle_stateful_request..run_serverz)Bad Request: No valid session ID provided) status_code)r=r>r'r5)rheadersgetrrr+rKr8rrhexrrrrr:r,rrLr rMr r BAD_REQUEST) r#r2r3r4requestrequest_mcp_session_id transportnew_session_idrSresponser$rHr%r7s>      .5z5StreamableHTTPSessionManager._handle_stateful_request)NFFN) rrrrrrrrrr)r'r()r2r r3r r4r r'r5) __name__ __module__ __qualname____doc__r& contextlibasynccontextmanagerr1r8r6r7r$r$r$r%rs   ( 1r)%ra __future__rrbloggingcollections.abcrhttprtypingruuidrr anyio.abcrstarlette.requestsrstarlette.responsesr starlette.typesr r r mcp.server.lowlevel.serverr MCPServermcp.server.streamable_httprrrmcp.server.transport_securityr getLoggerr^r+rr$r$r$r%s$