Fix mypy issues in nested handoff helper

Author: jhills20

src/agents/extensions/handoff_filters.py

@@ -2,7 +2,7 @@
 
 import json
 from copy import deepcopy
-from typing import Any
+from typing import Any, cast
 
 from ..handoffs import HandoffInputData
 from ..items import (
@@ -88,17 +88,19 @@ def _build_developer_message(transcript: list[TResponseInputItem]) -> TResponseI
     transcript_copy = [deepcopy(item) for item in transcript]
     if transcript_copy:
         summary_lines = [
-            f"{idx + 1}. {_format_transcript_item(item)}" for idx, item in enumerate(transcript_copy)
+            f"{idx + 1}. {_format_transcript_item(item)}"
+            for idx, item in enumerate(transcript_copy)
         ]
     else:
         summary_lines = ["(no previous turns recorded)"]
 
     content_lines = [_CONVERSATION_HISTORY_START, *summary_lines, _CONVERSATION_HISTORY_END]
     content = "\n".join(content_lines)
-    return {
+    developer_message: dict[str, Any] = {
         "role": "developer",
         "content": content,
     }
+    return cast(TResponseInputItem, developer_message)
 
 
 def _format_transcript_item(item: TResponseInputItem) -> str:
@@ -190,13 +192,13 @@ def _parse_summary_line(line: str) -> TResponseInputItem | None:
     if not role_text:
         return None
     role, name = _split_role_and_name(role_text)
-    reconstructed: TResponseInputItem = {"role": role}
+    reconstructed: dict[str, Any] = {"role": role}
     if name:
         reconstructed["name"] = name
     content = remainder.strip()
     if content:
         reconstructed["content"] = content
-    return reconstructed
+    return cast(TResponseInputItem, reconstructed)
 
 
 def _split_role_and_name(role_text: str) -> tuple[str, str | None]:

tests/test_agent_runner.py

@@ -43,6 +43,14 @@
 from .utils.simple_session import SimpleListSession
 
 
+def _as_message(item: Any) -> dict[str, Any]:
+    assert isinstance(item, dict)
+    role = item.get("role")
+    assert isinstance(role, str)
+    assert role in {"assistant", "user", "system", "developer"}
+    return cast(dict[str, Any], item)
+
+
 @pytest.mark.asyncio
 async def test_simple_first_run():
     model = FakeModel()
@@ -293,11 +301,15 @@ async def test_default_handoff_history_nested_and_filters_respected():
     result = await Runner.run(agent_2, input="user_message")
 
     assert isinstance(result.input, list)
-    assert result.input[0]["role"] == "developer"
-    assert "<CONVERSATION HISTORY>" in result.input[0]["content"]
-    assert "triage summary" in result.input[0]["content"]
-    assert result.input[1]["role"] == "user"
-    assert result.input[1]["content"] == "user_message"
+    developer = _as_message(result.input[0])
+    assert developer["role"] == "developer"
+    developer_content = developer["content"]
+    assert isinstance(developer_content, str)
+    assert "<CONVERSATION HISTORY>" in developer_content
+    assert "triage summary" in developer_content
+    latest_user = _as_message(result.input[1])
+    assert latest_user["role"] == "user"
+    assert latest_user["content"] == "user_message"
 
     passthrough_model = FakeModel()
     delegate = Agent(name="delegate", model=passthrough_model)
@@ -348,13 +360,16 @@ async def test_default_handoff_history_accumulates_across_multiple_handoffs():
     assert closer_model.first_turn_args is not None
     closer_input = closer_model.first_turn_args["input"]
     assert isinstance(closer_input, list)
-    assert closer_input[0]["role"] == "developer"
-    developer_content = closer_input[0]["content"]
+    developer = _as_message(closer_input[0])
+    assert developer["role"] == "developer"
+    developer_content = developer["content"]
+    assert isinstance(developer_content, str)
     assert developer_content.count("<CONVERSATION HISTORY>") == 1
     assert "triage summary" in developer_content
     assert "delegate update" in developer_content
-    assert closer_input[1]["role"] == "user"
-    assert closer_input[1]["content"] == "user_question"
+    latest_user = _as_message(closer_input[1])
+    assert latest_user["role"] == "user"
+    assert latest_user["content"] == "user_question"
 
 
 @pytest.mark.asyncio

tests/test_extension_filters.py

@@ -1,3 +1,5 @@
+from typing import Any, cast
+
 from openai.types.responses import ResponseOutputMessage, ResponseOutputText
 from openai.types.responses.response_reasoning_item import ResponseReasoningItem
 
@@ -96,6 +98,14 @@ def _get_reasoning_output_run_item() -> ReasoningItem:
     )
 
 
+def _as_message(item: TResponseInputItem) -> dict[str, Any]:
+    assert isinstance(item, dict)
+    role = item.get("role")
+    assert isinstance(role, str)
+    assert role in {"assistant", "user", "system", "developer"}
+    return cast(dict[str, Any], item)
+
+
 def test_empty_data():
     handoff_input_data = HandoffInputData(
         input_history=(),
@@ -242,13 +252,16 @@ def test_nest_handoff_history_wraps_transcript() -> None:
     nested = nest_handoff_history(data)
 
     assert isinstance(nested.input_history, tuple)
-    assert nested.input_history[0]["role"] == "developer"
-    developer_content = nested.input_history[0]["content"]
+    developer = _as_message(nested.input_history[0])
+    assert developer["role"] == "developer"
+    developer_content = developer["content"]
+    assert isinstance(developer_content, str)
     assert "<CONVERSATION HISTORY>" in developer_content
     assert "</CONVERSATION HISTORY>" in developer_content
     assert "Assist reply" in developer_content
-    assert nested.input_history[1]["role"] == "user"
-    assert nested.input_history[1]["content"] == "Hello"
+    latest_user = _as_message(nested.input_history[1])
+    assert latest_user["role"] == "user"
+    assert latest_user["content"] == "Hello"
     assert len(nested.pre_handoff_items) == 0
     assert nested.new_items == data.new_items
 
@@ -265,8 +278,11 @@ def test_nest_handoff_history_handles_missing_user() -> None:
 
     assert isinstance(nested.input_history, tuple)
     assert len(nested.input_history) == 1
-    assert nested.input_history[0]["role"] == "developer"
-    assert "reasoning" in nested.input_history[0]["content"].lower()
+    developer = _as_message(nested.input_history[0])
+    assert developer["role"] == "developer"
+    developer_content = developer["content"]
+    assert isinstance(developer_content, str)
+    assert "reasoning" in developer_content.lower()
 
 
 def test_nest_handoff_history_appends_existing_history() -> None:
@@ -278,9 +294,10 @@ def test_nest_handoff_history_appends_existing_history() -> None:
     )
 
     first_nested = nest_handoff_history(first)
+    assert isinstance(first_nested.input_history, tuple)
     developer_message = first_nested.input_history[0]
 
-    follow_up_history = (
+    follow_up_history: tuple[TResponseInputItem, ...] = (
         developer_message,
         _get_user_input_item("Another question"),
     )
@@ -295,9 +312,10 @@ def test_nest_handoff_history_appends_existing_history() -> None:
     second_nested = nest_handoff_history(second)
 
     assert isinstance(second_nested.input_history, tuple)
-    developer = second_nested.input_history[0]
+    developer = _as_message(second_nested.input_history[0])
     assert developer["role"] == "developer"
     content = developer["content"]
+    assert isinstance(content, str)
     assert content.count("<CONVERSATION HISTORY>") == 1
     assert content.count("</CONVERSATION HISTORY>") == 1
     assert "First reply" in content