Optionize initialized notification tolerance

Co-authored-by: Codex <noreply@openai.com>

Author: elainegan-openai

src/agents/mcp/server.py

@@ -79,6 +79,17 @@ class RequireApprovalObject(TypedDict, total=False):
 
 
 class _AgentsStreamableHTTPTransport(StreamableHTTPTransport):
+    def __init__(
+        self,
+        url: str,
+        *,
+        ignore_initialized_notification_failure: bool = False,
+    ) -> None:
+        super().__init__(url)
+        self._ignore_initialized_notification_failure = (
+            ignore_initialized_notification_failure
+        )
+
     async def post_writer(
         self,
         client: httpx.AsyncClient,
@@ -127,7 +138,10 @@ async def handle_request_async(
                         try:
                             await handle_request_async()
                         except Exception:
-                            if self._is_initialized_notification(message):
+                            if (
+                                self._ignore_initialized_notification_failure
+                                and self._is_initialized_notification(message)
+                            ):
                                 logger.warning(
                                     "Ignoring initialized notification failure in post_writer",
                                     exc_info=True,
@@ -150,6 +164,7 @@ async def streamablehttp_client(
     terminate_on_close: bool = True,
     httpx_client_factory: HttpClientFactory = create_mcp_http_client,
     auth: httpx.Auth | None = None,
+    ignore_initialized_notification_failure: bool = False,
 ):
     timeout_seconds = timeout.total_seconds() if isinstance(timeout, timedelta) else timeout
     sse_read_timeout_seconds = (
@@ -163,7 +178,10 @@ async def streamablehttp_client(
         timeout=httpx.Timeout(timeout_seconds, read=sse_read_timeout_seconds),
         auth=auth,
     )
-    transport = _AgentsStreamableHTTPTransport(url)
+    transport = _AgentsStreamableHTTPTransport(
+        url,
+        ignore_initialized_notification_failure=ignore_initialized_notification_failure,
+    )
 
     async with client:
         read_stream_writer, read_stream = anyio.create_memory_object_stream[
@@ -1286,6 +1304,14 @@ class MCPServerStreamableHttpParams(TypedDict):
     transport.
     """
 
+    ignore_initialized_notification_failure: NotRequired[bool]
+    """Whether to ignore failures when sending the best-effort
+    ``notifications/initialized`` POST.
+
+    Defaults to ``False``. When set to ``True``, initialized-notification failures are
+    logged and ignored so subsequent requests on the same transport can continue.
+    """
+
 
 class MCPServerStreamableHttp(_MCPServerWithClientSession):
     """MCP server implementation that uses the Streamable HTTP transport. See the [spec]
@@ -1380,6 +1406,10 @@ def create_streams(
             kwargs["httpx_client_factory"] = self.params["httpx_client_factory"]
         if "auth" in self.params:
             kwargs["auth"] = self.params["auth"]
+        if "ignore_initialized_notification_failure" in self.params:
+            kwargs["ignore_initialized_notification_failure"] = self.params[
+                "ignore_initialized_notification_failure"
+            ]
         return streamablehttp_client(**kwargs)
 
     @asynccontextmanager

tests/mcp/test_streamable_http_client_factory.py

@@ -256,7 +256,10 @@ def comprehensive_factory(
 
 @pytest.mark.asyncio
 async def test_initialized_notification_failure_does_not_stop_following_requests():
-    transport = _AgentsStreamableHTTPTransport("https://example.test/mcp")
+    transport = _AgentsStreamableHTTPTransport(
+        "https://example.test/mcp",
+        ignore_initialized_notification_failure=True,
+    )
     request_handled = asyncio.Event()
 
     async def fake_handle_post_request(ctx):
@@ -299,3 +302,27 @@ async def fake_handle_post_request(ctx):
 
             await write_stream.aclose()
             tg.cancel_scope.cancel()
+
+
+@pytest.mark.asyncio
+async def test_streamable_http_server_passes_ignore_initialized_notification_failure():
+    with patch("agents.mcp.server.streamablehttp_client") as mock_client:
+        mock_client.return_value = MagicMock()
+
+        server = MCPServerStreamableHttp(
+            params={
+                "url": "http://localhost:8000/mcp",
+                "ignore_initialized_notification_failure": True,
+            }
+        )
+
+        server.create_streams()
+
+        mock_client.assert_called_once_with(
+            url="http://localhost:8000/mcp",
+            headers=None,
+            timeout=5,
+            sse_read_timeout=300,
+            terminate_on_close=True,
+            ignore_initialized_notification_failure=True,
+        )