Optionize streamed Responses read-timeout policy

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

src/agents/mcp/server.py

@@ -10,6 +10,7 @@ from datetime import timedelta
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Callable, Literal, TypeVar, Union, cast
 
+import anyio
 import httpx
 
 if sys.version_info < (3, 11):
@@ -19,13 +20,19 @@ from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStre
 from mcp import ClientSession, StdioServerParameters, Tool as MCPTool, stdio_client
 from mcp.client.session import MessageHandlerFnT
 from mcp.client.sse import sse_client
-from mcp.client.streamable_http import GetSessionIdCallback, streamablehttp_client
+from mcp.client.streamable_http import (
+    GetSessionIdCallback,
+    RequestContext,
+    StreamableHTTPTransport,
+    create_mcp_http_client,
+)
 from mcp.shared.exceptions import McpError
-from mcp.shared.message import SessionMessage
+from mcp.shared.message import ClientMessageMetadata, SessionMessage
 from mcp.types import (
     CallToolResult,
     GetPromptResult,
     InitializeResult,
+    JSONRPCRequest,
     ListPromptsResult,
     ListResourcesResult,
     ListResourceTemplatesResult,
@@ -71,6 +78,143 @@ else:
 T = TypeVar("T")
 
 
+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,
+        write_stream_reader: MemoryObjectReceiveStream[SessionMessage],
+        read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception],
+        write_stream: MemoryObjectSendStream[SessionMessage],
+        start_get_stream: Callable[[], None],
+        tg: anyio.abc.TaskGroup,
+    ) -> None:
+        try:
+            async with write_stream_reader:
+                async for session_message in write_stream_reader:
+                    message = session_message.message
+                    metadata = (
+                        session_message.metadata
+                        if isinstance(session_message.metadata, ClientMessageMetadata)
+                        else None
+                    )
+                    is_resumption = bool(metadata and metadata.resumption_token)
+
+                    logger.debug(f"Sending client message: {message}")
+
+                    if self._is_initialized_notification(message):
+                        start_get_stream()
+
+                    ctx = RequestContext(
+                        client=client,
+                        session_id=self.session_id,
+                        session_message=session_message,
+                        metadata=metadata,
+                        read_stream_writer=read_stream_writer,
+                    )
+
+                    async def handle_request_async(
+                        request_ctx=ctx,
+                        request_is_resumption=is_resumption,
+                    ):
+                        if request_is_resumption:
+                            await self._handle_resumption_request(request_ctx)
+                        else:
+                            await self._handle_post_request(request_ctx)
+
+                    if isinstance(message.root, JSONRPCRequest):
+                        tg.start_soon(handle_request_async)
+                    else:
+                        try:
+                            await handle_request_async()
+                        except Exception:
+                            if (
+                                self._ignore_initialized_notification_failure
+                                and self._is_initialized_notification(message)
+                            ):
+                                logger.warning(
+                                    "Ignoring initialized notification failure in post_writer",
+                                    exc_info=True,
+                                )
+                                continue
+                            raise
+        except Exception:
+            logger.exception("Error in post_writer")  # pragma: no cover
+        finally:
+            await read_stream_writer.aclose()
+            await write_stream.aclose()
+
+
+@asynccontextmanager
+async def streamablehttp_client(
+    url: str,
+    headers: dict[str, str] | None = None,
+    timeout: float | timedelta = 30,
+    sse_read_timeout: float | timedelta = 60 * 5,
+    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 = (
+        sse_read_timeout.total_seconds()
+        if isinstance(sse_read_timeout, timedelta)
+        else sse_read_timeout
+    )
+
+    client = httpx_client_factory(
+        headers=headers,
+        timeout=httpx.Timeout(timeout_seconds, read=sse_read_timeout_seconds),
+        auth=auth,
+    )
+    transport = _AgentsStreamableHTTPTransport(
+        url,
+        ignore_initialized_notification_failure=ignore_initialized_notification_failure,
+    )
+
+    async with client:
+        read_stream_writer, read_stream = anyio.create_memory_object_stream[
+            SessionMessage | Exception
+        ](0)
+        write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)
+
+        async with anyio.create_task_group() as tg:
+            try:
+                def start_get_stream() -> None:
+                    tg.start_soon(transport.handle_get_stream, client, read_stream_writer)
+
+                tg.start_soon(
+                    transport.post_writer,
+                    client,
+                    write_stream_reader,
+                    read_stream_writer,
+                    write_stream,
+                    start_get_stream,
+                    tg,
+                )
+
+                try:
+                    yield (read_stream, write_stream, transport.get_session_id)
+                finally:
+                    if transport.session_id and terminate_on_close:
+                        await transport.terminate_session(client)
+                    tg.cancel_scope.cancel()
+            finally:
+                await read_stream_writer.aclose()
+                await write_stream.aclose()
+
+
 class _SharedSessionRequestNeedsIsolation(Exception):
     """Raised when a shared-session request should be retried on an isolated session."""
 
@@ -1160,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]
@@ -1254,6 +1406,10 @@ class MCPServerStreamableHttp(_MCPServerWithClientSession):
             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

src/agents/model_settings.py

@@ -155,6 +155,14 @@ class ModelSettings:
     """Additional headers to provide with the request.
     Defaults to None if not provided."""
 
+    disable_stream_read_timeout: bool | None = None
+    """Whether streamed Responses HTTP requests should disable the transport read deadline.
+
+    When set to ``True``, streamed Responses requests keep connect/write/pool timeouts but
+    remove the read timeout so long-lived SSE streams are not interrupted by transport-level
+    read deadlines. Defaults to ``None`` and preserves the SDK default behavior.
+    """
+
     extra_args: dict[str, Any] | None = None
     """Arbitrary keyword arguments to pass to the model API call.
     These will be passed directly to the underlying model provider's API.

src/agents/models/openai_responses.py

@@ -4,6 +4,7 @@ import asyncio
 import contextlib
 import inspect
 import json
+import time
 import weakref
 from collections.abc import AsyncIterator, Awaitable, Callable, Mapping, Sequence
 from contextvars import ContextVar
@@ -188,6 +189,39 @@ class _WebsocketRequestTimeouts:
     recv: float | None
 
 
+def _extract_httpx_read_timeout(timeout: Any) -> float | None:
+    if timeout is None or _is_openai_omitted_value(timeout):
+        return None
+
+    if isinstance(timeout, httpx.Timeout):
+        return None if timeout.read is None else float(timeout.read)
+
+    if isinstance(timeout, (int, float)):
+        return float(timeout)
+
+    return None
+
+
+def _streaming_timeout_without_read_deadline(timeout: Any) -> Any:
+    if timeout is None or _is_openai_omitted_value(timeout):
+        return timeout
+
+    if isinstance(timeout, httpx.Timeout):
+        if timeout.read is None:
+            return timeout
+        return httpx.Timeout(
+            connect=timeout.connect,
+            read=None,
+            write=timeout.write,
+            pool=timeout.pool,
+        )
+
+    if isinstance(timeout, (int, float)):
+        return httpx.Timeout(float(timeout), read=None)
+
+    return timeout
+
+
 class _ResponseStreamWithRequestId:
     """Wrap an SDK event stream and retain the originating request ID."""
 
@@ -203,15 +237,20 @@ class _ResponseStreamWithRequestId:
         stream: AsyncIterator[ResponseStreamEvent],
         *,
         request_id: str | None,
+        read_timeout_seconds: float | None,
         cleanup: Callable[[], Awaitable[object]],
     ) -> None:
         self._stream = stream
         self.request_id = request_id
+        self._read_timeout_seconds = read_timeout_seconds
         self._cleanup = cleanup
         self._closed = False
         self._stream_close_complete = False
         self._cleanup_complete = False
         self._yielded_terminal_event = False
+        self._event_count = 0
+        self._last_event_type: str | None = None
+        self._last_event_monotonic: float | None = None
 
     def __aiter__(self) -> _ResponseStreamWithRequestId:
         return self
@@ -222,6 +261,26 @@ class _ResponseStreamWithRequestId:
 
         try:
             event = await self._stream.__anext__()
+        except httpx.ReadTimeout:
+            seconds_since_last_event = (
+                None
+                if self._last_event_monotonic is None
+                else time.monotonic() - self._last_event_monotonic
+            )
+            logger.warning(
+                "responses_stream_read_timeout "
+                "request_id=%s "
+                "read_timeout_seconds=%s "
+                "event_count=%s "
+                "last_event_type=%s "
+                "seconds_since_last_event=%s",
+                self.request_id,
+                self._read_timeout_seconds,
+                self._event_count,
+                self._last_event_type,
+                seconds_since_last_event,
+            )
+            raise
         except StopAsyncIteration:
             self._closed = True
             await self._cleanup_after_exhaustion()
@@ -229,6 +288,9 @@ class _ResponseStreamWithRequestId:
 
         self._attach_request_id(event)
         event_type = getattr(event, "type", None)
+        self._event_count += 1
+        self._last_event_type = event_type if isinstance(event_type, str) else None
+        self._last_event_monotonic = time.monotonic()
         if event_type in self._TERMINAL_EVENT_TYPES:
             self._yielded_terminal_event = True
         return event
@@ -653,6 +715,14 @@ class OpenAIResponsesModel(Model):
 
         # Keep the raw API response open while callers consume the SSE stream so we can expose
         # its request ID on terminal response payloads before cleanup closes the transport.
+        request_timeout = create_kwargs.get("timeout", omit)
+        if _is_openai_omitted_value(request_timeout):
+            request_timeout = getattr(client, "timeout", None)
+        if model_settings.disable_stream_read_timeout:
+            stream_request_timeout = _streaming_timeout_without_read_deadline(request_timeout)
+            if not _is_openai_omitted_value(stream_request_timeout):
+                create_kwargs = dict(create_kwargs)
+                create_kwargs["timeout"] = stream_request_timeout
         api_response_cm = stream_create(**create_kwargs)
         api_response = await api_response_cm.__aenter__()
         try:
@@ -664,6 +734,7 @@ class OpenAIResponsesModel(Model):
         return _ResponseStreamWithRequestId(
             cast(AsyncIterator[ResponseStreamEvent], stream_response),
             request_id=getattr(api_response, "request_id", None),
+            read_timeout_seconds=_extract_httpx_read_timeout(request_timeout),
             cleanup=lambda: api_response_cm.__aexit__(None, None, None),
         )
 

src/agents/run_internal/tool_execution.py

@@ -1400,9 +1400,10 @@ class _FunctionToolBatchExecutor:
         self,
         tasks: set[asyncio.Task[Any]],
     ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
-        late_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = {
-            task: "cancelled_teardown" for task in tasks
-        }
+        late_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = dict.fromkeys(
+            tasks,
+            "cancelled_teardown",
+        )
         return await _drain_cancelled_function_tool_tasks(
             pending_tasks=tasks,
             task_states=self.task_states,
@@ -1415,9 +1416,9 @@ class _FunctionToolBatchExecutor:
         self,
         tasks: set[asyncio.Task[Any]],
     ) -> tuple[_FunctionToolFailure | None, set[asyncio.Task[Any]]]:
-        post_invoke_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = {
-            task: "post_invoke" for task in tasks
-        }
+        post_invoke_failure_sources: dict[asyncio.Task[Any], _FunctionToolFailureSource] = (
+            dict.fromkeys(tasks, "post_invoke")
+        )
         return await _wait_pending_function_tool_tasks_for_timeout(
             pending_tasks=tasks,
             task_states=self.task_states,
@@ -1638,7 +1639,7 @@ class _FunctionToolBatchExecutor:
                 arguments=tool_call.arguments,
             )
         except asyncio.CancelledError as e:
-            if not self.isolate_parallel_failures or outer_task in self.teardown_cancelled_tasks:
+            if outer_task in self.teardown_cancelled_tasks:
                 raise
 
             result = await maybe_invoke_function_tool_failure_error_function(

tests/mcp/test_streamable_http_client_factory.py

@@ -2,12 +2,17 @@
 
 from __future__ import annotations
 
+import asyncio
 from unittest.mock import MagicMock, patch
 
+import anyio
 import httpx
 import pytest
+from mcp.shared.message import JSONRPCMessage, SessionMessage
+from mcp.types import JSONRPCNotification, JSONRPCRequest
 
 from agents.mcp import MCPServerStreamableHttp
+from agents.mcp.server import _AgentsStreamableHTTPTransport
 
 
 class TestMCPServerStreamableHttpClientFactory:
@@ -247,3 +252,77 @@ class TestMCPServerStreamableHttpClientFactory:
                 terminate_on_close=False,
                 httpx_client_factory=comprehensive_factory,
             )
+
+
+@pytest.mark.asyncio
+async def test_initialized_notification_failure_does_not_stop_following_requests():
+    transport = _AgentsStreamableHTTPTransport(
+        "https://example.test/mcp",
+        ignore_initialized_notification_failure=True,
+    )
+    request_handled = asyncio.Event()
+
+    async def fake_handle_post_request(ctx):
+        message = ctx.session_message.message
+        if transport._is_initialized_notification(message):
+            request = httpx.Request("POST", "https://example.test/mcp")
+            response = httpx.Response(503, request=request)
+            raise httpx.HTTPStatusError("HTTP error 503", request=request, response=response)
+        request_handled.set()
+
+    transport._handle_post_request = fake_handle_post_request  # type: ignore[method-assign]
+
+    read_stream_writer, _ = anyio.create_memory_object_stream[SessionMessage | Exception](0)
+    write_stream, write_stream_reader = anyio.create_memory_object_stream[SessionMessage](0)
+
+    initialized_notification = SessionMessage(
+        JSONRPCMessage(
+            JSONRPCNotification(jsonrpc="2.0", method="notifications/initialized", params={})
+        )
+    )
+    list_tools_request = SessionMessage(
+        JSONRPCMessage(JSONRPCRequest(jsonrpc="2.0", id=1, method="tools/list", params={}))
+    )
+
+    async with httpx.AsyncClient() as client:
+        async with anyio.create_task_group() as tg:
+            tg.start_soon(
+                transport.post_writer,
+                client,
+                write_stream_reader,
+                read_stream_writer,
+                write_stream,
+                lambda: None,
+                tg,
+            )
+            await write_stream.send(initialized_notification)
+            await write_stream.send(list_tools_request)
+
+            await asyncio.wait_for(request_handled.wait(), timeout=1)
+
+            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,
+        )

tests/model_settings/test_serialization.py

@@ -65,6 +65,7 @@ def test_all_fields_serialization() -> None:
         extra_query={"foo": "bar"},
         extra_body={"foo": "bar"},
         extra_headers={"foo": "bar"},
+        disable_stream_read_timeout=True,
         extra_args={"custom_param": "value", "another_param": 42},
         retry=ModelRetrySettings(
             max_retries=2,
@@ -181,6 +182,7 @@ def test_pydantic_serialization() -> None:
         extra_query={"foo": "bar"},
         extra_body={"foo": "bar"},
         extra_headers={"foo": "bar"},
+        disable_stream_read_timeout=True,
         extra_args={"custom_param": "value", "another_param": 42},
     )
 

tests/test_agent_runner.py

@@ -641,6 +641,45 @@ async def test_parallel_tool_call_with_cancelled_sibling_reaches_final_output()
     ]
 
 
+@pytest.mark.asyncio
+async def test_single_tool_call_with_cancelled_tool_reaches_final_output() -> None:
+    async def _cancel_tool() -> str:
+        raise asyncio.CancelledError("tool-cancelled")
+
+    model = FakeModel()
+    agent = Agent(
+        name="test",
+        model=model,
+        tools=[function_tool(_cancel_tool, name_override="cancel_tool")],
+    )
+
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("cancel_tool", "{}", call_id="call_cancel")],
+            [get_text_message("final answer")],
+        ]
+    )
+
+    result = await Runner.run(agent, input="user_message")
+
+    assert result.final_output == "final answer"
+    assert len(result.raw_responses) == 2
+
+    second_turn_input = cast(list[dict[str, Any]], model.last_turn_args["input"])
+    tool_outputs = [
+        item for item in second_turn_input if item.get("type") == "function_call_output"
+    ]
+    assert tool_outputs == [
+        {
+            "call_id": "call_cancel",
+            "output": (
+                "An error occurred while running the tool. Please try again. Error: tool-cancelled"
+            ),
+            "type": "function_call_output",
+        },
+    ]
+
+
 @pytest.mark.asyncio
 async def test_reasoning_item_id_policy_omits_follow_up_reasoning_ids() -> None:
     model = FakeModel()

tests/test_agent_runner_streamed.py

@@ -496,6 +496,46 @@ async def test_streamed_parallel_tool_call_with_cancelled_sibling_reaches_final_
     ]
 
 
+@pytest.mark.asyncio
+async def test_streamed_single_tool_call_with_cancelled_tool_reaches_final_output() -> None:
+    async def _cancel_tool() -> str:
+        raise asyncio.CancelledError("tool-cancelled")
+
+    model = FakeModel()
+    agent = Agent(
+        name="test",
+        model=model,
+        tools=[function_tool(_cancel_tool, name_override="cancel_tool")],
+    )
+
+    model.add_multiple_turn_outputs(
+        [
+            [get_function_tool_call("cancel_tool", "{}", call_id="call_cancel")],
+            [get_text_message("final answer")],
+        ]
+    )
+
+    result = Runner.run_streamed(agent, input="user_message")
+    await consume_stream(result)
+
+    assert result.final_output == "final answer"
+    assert len(result.raw_responses) == 2
+
+    second_turn_input = cast(list[dict[str, Any]], model.last_turn_args["input"])
+    tool_outputs = [
+        item for item in second_turn_input if item.get("type") == "function_call_output"
+    ]
+    assert tool_outputs == [
+        {
+            "call_id": "call_cancel",
+            "output": (
+                "An error occurred while running the tool. Please try again. Error: tool-cancelled"
+            ),
+            "type": "function_call_output",
+        },
+    ]
+
+
 @pytest.mark.asyncio
 async def test_streamed_reasoning_item_id_policy_omits_follow_up_reasoning_ids() -> None:
     model = FakeModel()

tests/test_openai_responses.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import itertools
 import json
 from types import SimpleNamespace
 from typing import Any, cast
@@ -32,6 +33,7 @@ from agents.models.openai_responses import (
     OpenAIResponsesModel,
     OpenAIResponsesWSModel,
     ResponsesWebSocketError,
+    _ResponseStreamWithRequestId,
     _should_retry_pre_event_websocket_disconnect,
 )
 from agents.retry import ModelRetryAdviceRequest
@@ -109,6 +111,50 @@ def _connection_closed_error(message: str) -> Exception:
     return ConnectionClosedError(message)
 
 
+class _ReadTimeoutAfterOneEventStream:
+    def __init__(self):
+        self._yielded = False
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._yielded:
+            self._yielded = True
+            return SimpleNamespace(type="response.in_progress", response=SimpleNamespace())
+        raise httpx.ReadTimeout("timed out")
+
+
+@pytest.mark.asyncio
+async def test_response_stream_with_request_id_logs_read_timeout_context(caplog, monkeypatch):
+    monotonic_values = itertools.chain([100.0, 700.5], itertools.repeat(700.5))
+    monkeypatch.setattr(
+        "agents.models.openai_responses.time.monotonic",
+        lambda: next(monotonic_values),
+    )
+
+    stream = _ResponseStreamWithRequestId(
+        _ReadTimeoutAfterOneEventStream(),
+        request_id="req_123",
+        read_timeout_seconds=600.0,
+        cleanup=lambda: asyncio.sleep(0),
+    )
+
+    first_event = await stream.__anext__()
+    assert first_event.type == "response.in_progress"
+
+    with caplog.at_level("WARNING"):
+        with pytest.raises(httpx.ReadTimeout):
+            await stream.__anext__()
+
+    assert "responses_stream_read_timeout" in caplog.text
+    assert "request_id=req_123" in caplog.text
+    assert "read_timeout_seconds=600.0" in caplog.text
+    assert "event_count=1" in caplog.text
+    assert "last_event_type=response.in_progress" in caplog.text
+    assert "seconds_since_last_event=600.5" in caplog.text
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 @pytest.mark.parametrize("override_ua", [None, "test_user_agent"])
@@ -299,6 +345,199 @@ async def test_fetch_response_stream_attaches_request_id_to_terminal_response():
     assert aexit_calls == [(None, None, None)]
 
 
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+async def test_stream_read_timeout_disable_uses_client_timeout():
+    class DummyHTTPStream:
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            raise StopAsyncIteration
+
+    inner_stream = DummyHTTPStream()
+
+    class DummyAPIResponse:
+        request_id = "req_stream_timeout_client"
+
+        async def parse(self):
+            return inner_stream
+
+    api_response = DummyAPIResponse()
+    captured_kwargs: dict[str, Any] = {}
+
+    class DummyStreamingContextManager:
+        async def __aenter__(self):
+            return api_response
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+    class DummyResponses:
+        def __init__(self):
+            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)
+
+        def create_streaming(self, **kwargs):
+            captured_kwargs.update(kwargs)
+            return DummyStreamingContextManager()
+
+    class DummyResponsesClient:
+        def __init__(self):
+            self.responses = DummyResponses()
+            self.timeout = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0)
+
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]
+
+    stream = await model._fetch_response(
+        system_instructions=None,
+        input="hi",
+        model_settings=ModelSettings(disable_stream_read_timeout=True),
+        tools=[],
+        output_schema=None,
+        handoffs=[],
+        previous_response_id=None,
+        conversation_id=None,
+        stream=True,
+    )
+
+    with pytest.raises(StopAsyncIteration):
+        await cast(Any, stream).__anext__()
+
+    timeout = captured_kwargs["timeout"]
+    assert isinstance(timeout, httpx.Timeout)
+    assert timeout.connect == 5.0
+    assert timeout.read is None
+    assert timeout.write == 600.0
+    assert timeout.pool == 600.0
+
+
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+async def test_stream_read_timeout_disable_uses_override_timeout():
+    class DummyHTTPStream:
+        def __aiter__(self):
+            return self
+
+        async def __anext__(self):
+            raise StopAsyncIteration
+
+    inner_stream = DummyHTTPStream()
+
+    class DummyAPIResponse:
+        request_id = "req_stream_timeout_override"
+
+        async def parse(self):
+            return inner_stream
+
+    api_response = DummyAPIResponse()
+    captured_kwargs: dict[str, Any] = {}
+
+    class DummyStreamingContextManager:
+        async def __aenter__(self):
+            return api_response
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+    class DummyResponses:
+        def __init__(self):
+            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)
+
+        def create_streaming(self, **kwargs):
+            captured_kwargs.update(kwargs)
+            return DummyStreamingContextManager()
+
+    class DummyResponsesClient:
+        def __init__(self):
+            self.responses = DummyResponses()
+            self.timeout = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0)
+
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]
+
+    stream = await model._fetch_response(
+        system_instructions=None,
+        input="hi",
+        model_settings=ModelSettings(
+            disable_stream_read_timeout=True,
+            extra_args={"timeout": 30.0},
+        ),
+        tools=[],
+        output_schema=None,
+        handoffs=[],
+        previous_response_id=None,
+        conversation_id=None,
+        stream=True,
+    )
+
+    with pytest.raises(StopAsyncIteration):
+        await cast(Any, stream).__anext__()
+
+    timeout = captured_kwargs["timeout"]
+    assert isinstance(timeout, httpx.Timeout)
+    assert timeout.connect == 30.0
+    assert timeout.read is None
+    assert timeout.write == 30.0
+    assert timeout.pool == 30.0
+
+
+@pytest.mark.allow_call_model_methods
+@pytest.mark.asyncio
+async def test_fetch_response_stream_preserves_timeout_by_default():
+    class DummyInnerStream:
+        async def __anext__(self):
+            raise StopAsyncIteration
+
+    inner_stream = DummyInnerStream()
+
+    class DummyAPIResponse:
+        request_id = "req_stream_123"
+
+        async def parse(self):
+            return inner_stream
+
+    api_response = DummyAPIResponse()
+    captured_kwargs: dict[str, Any] = {}
+
+    class DummyStreamingContextManager:
+        async def __aenter__(self):
+            return api_response
+
+        async def __aexit__(self, exc_type, exc, tb):
+            return False
+
+    class DummyResponses:
+        def __init__(self):
+            self.with_streaming_response = SimpleNamespace(create=self.create_streaming)
+
+        def create_streaming(self, **kwargs):
+            captured_kwargs.update(kwargs)
+            return DummyStreamingContextManager()
+
+    class DummyResponsesClient:
+        def __init__(self):
+            self.responses = DummyResponses()
+            self.timeout = httpx.Timeout(connect=5.0, read=600.0, write=600.0, pool=600.0)
+
+    model = OpenAIResponsesModel(model="gpt-4", openai_client=DummyResponsesClient())  # type: ignore[arg-type]
+
+    stream = await model._fetch_response(
+        system_instructions=None,
+        input="hi",
+        model_settings=ModelSettings(),
+        tools=[],
+        output_schema=None,
+        handoffs=[],
+        previous_response_id=None,
+        conversation_id=None,
+        stream=True,
+    )
+
+    with pytest.raises(StopAsyncIteration):
+        await cast(Any, stream).__anext__()
+
+    assert "timeout" not in captured_kwargs
+
+
 @pytest.mark.allow_call_model_methods
 @pytest.mark.asyncio
 async def test_fetch_response_stream_parse_failure_exits_streaming_context():

tests/test_run_step_execution.py

@@ -740,6 +740,29 @@ async def test_multiple_tool_calls_use_default_failure_error_function_for_manual
     )
 
 
+@pytest.mark.asyncio
+async def test_single_tool_call_uses_default_failure_error_function_for_cancelled_tool():
+    async def _cancel_tool() -> str:
+        raise asyncio.CancelledError("tool-cancelled")
+
+    cancel_tool = function_tool(_cancel_tool, name_override="cancel_tool")
+    agent = Agent(name="test", tools=[cancel_tool])
+    response = ModelResponse(
+        output=[get_function_tool_call("cancel_tool", "{}", call_id="1")],
+        usage=Usage(),
+        response_id=None,
+    )
+
+    result = await get_execute_result(agent, response)
+
+    assert len(result.generated_items) == 2
+    assert isinstance(result.next_step, NextStepRunAgain)
+    assert_item_is_function_tool_call_output(
+        result.generated_items[1],
+        "An error occurred while running the tool. Please try again. Error: tool-cancelled",
+    )
+
+
 @pytest.mark.asyncio
 async def test_multiple_tool_calls_surface_hook_failure_over_sibling_cancellation():
     hook_started = asyncio.Event()