Merge pull request #110 from IBM/streamable-http-support

Streamable http support

Author: crivetimihai

.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
 #####################################

README.md

@@ -30,7 +30,7 @@ It supports:
 
 * Federation across multiple MCP and REST services
 * Virtualization of legacy APIs as MCP-compliant tools and servers
-* Transport over HTTP, JSON-RPC, WebSocket, SSE, and stdio
+* Transport over HTTP, JSON-RPC, WebSocket, SSE, stdio and streamable-HTTP
 * An Admin UI for real-time management and configuration
 * Built-in auth, observability, retries, and rate-limiting
 * Scalable deployments via Docker or PyPI, Redis-backed caching, and multi-cluster federation
@@ -833,11 +833,13 @@ You can get started by copying the provided [.env.example](.env.example) to `.en
 
 ### Transport
 
-| Setting                   | Description            | Default | Options                         |
-| ------------------------- | ---------------------- | ------- | ------------------------------- |
-| `TRANSPORT_TYPE`          | Enabled transports     | `all`   | `http`,`ws`,`sse`,`stdio`,`all` |
-| `WEBSOCKET_PING_INTERVAL` | WebSocket ping (secs)  | `30`    | int > 0                         |
-| `SSE_RETRY_TIMEOUT`       | SSE retry timeout (ms) | `5000`  | int > 0                         |
+| Setting                   | Description                        | Default | Options                         |
+| ------------------------- | ---------------------------------- | ------- | ------------------------------- |
+| `TRANSPORT_TYPE`          | Enabled transports                 | `all`   | `http`,`ws`,`sse`,`stdio`,`all` |
+| `WEBSOCKET_PING_INTERVAL` | WebSocket ping (secs)              | `30`    | int > 0                         |
+| `SSE_RETRY_TIMEOUT`       | SSE retry timeout (ms)             | `5000`  | int > 0                         |
+| `USE_STATEFUL_SESSIONS`   | streamable http config             | `false` | bool                            |
+| `JSON_RESPONSE_ENABLED`   | json/sse streams (streamable http) | `true`  | bool                            |
 
 ### 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

@@ -104,6 +104,7 @@
     ToolService,
 )
 from mcpgateway.transports.sse_transport import SSETransport
+from mcpgateway.transports.streamablehttp_transport import SessionManagerWrapper, JWTAuthMiddlewareStreamableHttp
 from mcpgateway.types import (
     InitializeRequest,
     InitializeResult,
@@ -144,6 +145,9 @@
 sampling_handler = SamplingHandler()
 server_service = ServerService()
 
+# Initialize session manager for Streamable HTTP transport
+streamable_http_session = SessionManagerWrapper()
+
 
 # Initialize session registry
 session_registry = SessionRegistry(
@@ -190,13 +194,16 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
         await logging_service.initialize()
         await sampling_handler.initialize()
         await resource_cache.initialize()
+        await streamable_http_session.start()
+
         logger.info("All services initialized successfully")
         yield
     except Exception as e:
         logger.error(f"Error during startup: {str(e)}")
         raise
     finally:
         logger.info("Shutting down MCP Gateway services")
+        # await stop_streamablehttp()
         for service in [
             resource_cache,
             sampling_handler,
@@ -207,6 +214,7 @@ async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
             prompt_service,
             resource_service,
             tool_service,
+            streamable_http_session
         ]:
             try:
                 await service.shutdown()
@@ -276,6 +284,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
@@ -2028,6 +2039,9 @@ async def readiness_check(db: Session = Depends(get_db)):
 else:
     logger.warning("Admin API routes not mounted - Admin API disabled via MCPGATEWAY_ADMIN_API_ENABLED=False")
 
+# Streamable http Mount
+app.mount("/mcp", app=streamable_http_session.handle_streamable_http)
+
 # Conditional static files mounting and root redirect
 if UI_ENABLED:
     # Mount static files for UI