Apply review feedback

- Add Content-Type headers to test scopes to fix CI
- Add public is_terminated property to avoid accessing private _terminated
- Remove unnecessary try-catch from test
- Change logging from warning to error for session crashes

Author: felixweinberger

src/mcp/server/streamable_http.py

@@ -173,6 +173,11 @@ def __init__(
         ] = {}
         self._terminated = False
 
+    @property
+    def is_terminated(self) -> bool:
+        """Check if this transport has been explicitly terminated."""
+        return self._terminated
+
     def _create_error_response(
         self,
         error_message: str,
@@ -452,7 +457,7 @@ async def sse_writer():
                                 ):
                                     break
                     except Exception as e:
-                        logger.warning(f"Error in SSE writer: {e}", exc_info=True)
+                        logger.exception(f"Error in SSE writer: {e}")
                     finally:
                         logger.debug("Closing SSE writer")
                         await self._clean_up_memory_streams(request_id)
@@ -482,13 +487,13 @@ async def sse_writer():
                         session_message = SessionMessage(message, metadata=metadata)
                         await writer.send(session_message)
                 except Exception:
-                    logger.warning("SSE response error", exc_info=True)
+                    logger.exception("SSE response error")
                     await sse_stream_writer.aclose()
                     await sse_stream_reader.aclose()
                     await self._clean_up_memory_streams(request_id)
 
         except Exception as err:
-            logger.warning("Error handling POST request", exc_info=True)
+            logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",
                 HTTPStatus.INTERNAL_SERVER_ERROR,
@@ -570,7 +575,7 @@ async def standalone_sse_writer():
                         event_data = self._create_event_data(event_message)
                         await sse_stream_writer.send(event_data)
             except Exception as e:
-                logger.warning(f"Error in standalone SSE writer: {e}", exc_info=True)
+                logger.exception(f"Error in standalone SSE writer: {e}")
             finally:
                 logger.debug("Closing standalone SSE writer")
                 await self._clean_up_memory_streams(GET_STREAM_KEY)
@@ -586,7 +591,7 @@ async def standalone_sse_writer():
             # This will send headers immediately and establish the SSE connection
             await response(request.scope, request.receive, send)
         except Exception as e:
-            logger.warning(f"Error in standalone SSE response: {e}", exc_info=True)
+            logger.exception(f"Error in standalone SSE response: {e}")
             await sse_stream_writer.aclose()
             await sse_stream_reader.aclose()
             await self._clean_up_memory_streams(GET_STREAM_KEY)

src/mcp/server/streamable_http_manager.py

@@ -180,7 +180,7 @@ async def run_stateless_server(*, task_status: TaskStatus[None] = anyio.TASK_STA
                         stateless=True,
                     )
                 except Exception as e:
-                    logger.warning(f"Stateless session crashed: {e}", exc_info=True)
+                    logger.error(f"Stateless session crashed: {e}", exc_info=True)
 
         # Assert task group is not None for type checking
         assert self._task_group is not None
@@ -243,7 +243,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                                 stateless=False,  # Stateful mode
                             )
                         except Exception as e:
-                            logger.warning(
+                            logger.error(
                                 f"Session {http_transport.mcp_session_id} crashed: {e}",
                                 exc_info=True,
                             )
@@ -252,9 +252,7 @@ async def run_server(*, task_status: TaskStatus[None] = anyio.TASK_STATUS_IGNORE
                             if (
                                 http_transport.mcp_session_id
                                 and http_transport.mcp_session_id in self._server_instances
-                                and not (
-                                    hasattr(http_transport, "_terminated") and http_transport._terminated  # pyright: ignore
-                                )
+                                and not http_transport.is_terminated
                             ):
                                 logger.info(
                                     "Cleaning up crashed session "

tests/server/test_streamable_http_manager.py

@@ -104,7 +104,12 @@ async def test_stateful_session_cleanup_on_graceful_exit(running_manager):
     async def mock_send(message):
         sent_messages.append(message)
 
-    scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []}
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [(b"content-type", b"application/json")],
+    }
 
     async def mock_receive():
         return {"type": "http.request", "body": b"", "more_body": False}
@@ -135,9 +140,9 @@ async def mock_receive():
     # Give other tasks a chance to run. This is important for the finally block.
     await anyio.sleep(0.01)
 
-    assert (
-        session_id not in manager._server_instances
-    ), "Session ID should be removed from _server_instances after graceful exit"
+    assert session_id not in manager._server_instances, (
+        "Session ID should be removed from _server_instances after graceful exit"
+    )
     assert not manager._server_instances, "No sessions should be tracked after the only session exits gracefully"
 
 
@@ -158,21 +163,18 @@ async def mock_send(message):
         if message["type"] == "http.response.start" and message["status"] >= 500:
             pass  # Expected if TestException propagates that far up the transport
 
-    scope = {"type": "http", "method": "POST", "path": "/mcp", "headers": []}
+    scope = {
+        "type": "http",
+        "method": "POST",
+        "path": "/mcp",
+        "headers": [(b"content-type", b"application/json")],
+    }
 
     async def mock_receive():
         return {"type": "http.request", "body": b"", "more_body": False}
 
-    # It's possible handle_request itself might raise an error if the TestException
-    # isn't caught by the transport layer before propagating.
-    # The key is that the session manager's internal task for MCPServer.run
-    # encounters the exception.
-    try:
-        await manager.handle_request(scope, mock_receive, mock_send)
-    except TestException:
-        # This might be caught here if not handled by StreamableHTTPServerTransport's
-        # error handling
-        pass
+    # Trigger session creation
+    await manager.handle_request(scope, mock_receive, mock_send)
 
     session_id = None
     for msg in sent_messages:
@@ -191,7 +193,7 @@ async def mock_receive():
     # Give other tasks a chance to run to ensure the finally block executes
     await anyio.sleep(0.01)
 
-    assert (
-        session_id not in manager._server_instances
-    ), "Session ID should be removed from _server_instances after an exception"
+    assert session_id not in manager._server_instances, (
+        "Session ID should be removed from _server_instances after an exception"
+    )
     assert not manager._server_instances, "No sessions should be tracked after the only session crashes"