added authentication and stateful session support

Author: kevalmahajan

.env.example

@@ -140,6 +140,18 @@ WEBSOCKET_PING_INTERVAL=30
 # SSE client retry timeout (milliseconds)
 SSE_RETRY_TIMEOUT=5000
 
+
+#####################################
+# Streamabe HTTP Transport Configuration
+#####################################
+
+# Set False to use stateless sessions without event store and True for stateful sessions
+USE_STATEFUL_SESSIONS=false  
+
+# Set true for JSON responses, false for SSE streams
+JSON_RESPONSE_ENABLED=true 
+
+
 #####################################
 # Federation
 #####################################

mcpgateway/config.py

@@ -191,6 +191,10 @@ def _parse_federation_peers(cls, v):
     session_ttl: int = 3600
     message_ttl: int = 600
 
+    # streamable http transport
+    use_stateful_sessions: bool = False  # Set to False to use stateless sessions without event store
+    json_response_enabled: bool = True  # Enable JSON responses instead of SSE streams
+
     # Development
     dev_mode: bool = False
     reload: bool = False

mcpgateway/main.py

@@ -105,7 +105,7 @@
     ToolService,
 )
 from mcpgateway.transports.sse_transport import SSETransport
-from mcpgateway.transports.streamablehttp_transport import handle_streamable_http, session_manager
+from mcpgateway.transports.streamablehttp_transport import SessionManagerWrapper, JWTAuthMiddlewareStreamableHttp
 from mcpgateway.types import (
     InitializeRequest,
     InitializeResult,
@@ -146,6 +146,9 @@
 sampling_handler = SamplingHandler()
 server_service = ServerService()
 
+# Initialize session manager for Streamable HTTP transport
+streamable_http_session = SessionManagerWrapper()
+
 
 # Initialize session registry
 session_registry = SessionRegistry(
@@ -192,11 +195,8 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         await logging_service.initialize()
         await sampling_handler.initialize()
         await resource_cache.initialize()
+        await streamable_http_session.start()
 
-        from contextlib import AsyncExitStack
-        stack = AsyncExitStack()
-        await stack.enter_async_context(session_manager.run())
-        # await start_streamablehttp()
         logger.info("All services initialized successfully")
         yield
     except Exception as e:
@@ -215,6 +215,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
             prompt_service,
             resource_service,
             tool_service,
+            streamable_http_session
         ]:
             try:
                 await service.shutdown()
@@ -284,6 +285,9 @@ async def dispatch(self, request: Request, call_next):
 # Add custom DocsAuthMiddleware
 app.add_middleware(DocsAuthMiddleware)
 
+# Add streamable HTTP middleware for JWT auth
+app.add_middleware(JWTAuthMiddlewareStreamableHttp)
+
 # Set up Jinja2 templates and store in app state for later use
 templates = Jinja2Templates(directory=str(settings.templates_dir))
 app.state.templates = templates
@@ -2027,7 +2031,7 @@ async def readiness_check(db: Session = Depends(get_db)):
     logger.warning("Admin API routes not mounted - Admin API disabled via MCPGATEWAY_ADMIN_API_ENABLED=False")
 
 # Streamable http Mount
-app.mount("/mcp", app=handle_streamable_http)
+app.mount("/mcp", app=streamable_http_session.handle_streamable_http)
 
 # Conditional static files mounting and root redirect
 if UI_ENABLED: