Merge branch 'strands-agents:main' into message-requirements

Author: dbschmigelski

src/strands/event_loop/streaming.py

@@ -10,6 +10,7 @@
     ModelStopReason,
     ModelStreamChunkEvent,
     ModelStreamEvent,
+    ReasoningRedactedContentStreamEvent,
     ReasoningSignatureStreamEvent,
     ReasoningTextStreamEvent,
     TextStreamEvent,
@@ -170,6 +171,10 @@ def handle_content_block_delta(
                 delta=delta_content,
             )
 
+        elif redacted_content := delta_content["reasoningContent"].get("redactedContent"):
+            state["redactedContent"] = state.get("redactedContent", b"") + redacted_content
+            typed_event = ReasoningRedactedContentStreamEvent(redacted_content=redacted_content, delta=delta_content)
+
     return state, typed_event
 
 
@@ -188,6 +193,7 @@ def handle_content_block_stop(state: dict[str, Any]) -> dict[str, Any]:
     text = state["text"]
     reasoning_text = state["reasoningText"]
     citations_content = state["citationsContent"]
+    redacted_content = state.get("redactedContent")
 
     if current_tool_use:
         if "input" not in current_tool_use:
@@ -231,6 +237,9 @@ def handle_content_block_stop(state: dict[str, Any]) -> dict[str, Any]:
 
         content.append(content_block)
         state["reasoningText"] = ""
+    elif redacted_content:
+        content.append({"reasoningContent": {"redactedContent": redacted_content}})
+        state["redactedContent"] = b""
 
     return state
 

src/strands/models/openai.py

@@ -64,12 +64,10 @@ def __init__(self, client_args: Optional[dict[str, Any]] = None, **model_config:
         """
         validate_config_keys(model_config, self.OpenAIConfig)
         self.config = dict(model_config)
+        self.client_args = client_args or {}
 
         logger.debug("config=<%s> | initializing", self.config)
 
-        client_args = client_args or {}
-        self.client = openai.AsyncOpenAI(**client_args)
-
     @override
     def update_config(self, **model_config: Unpack[OpenAIConfig]) -> None:  # type: ignore[override]
         """Update the OpenAI model configuration with the provided arguments.
@@ -379,58 +377,60 @@ async def stream(
         logger.debug("formatted request=<%s>", request)
 
         logger.debug("invoking model")
-        response = await self.client.chat.completions.create(**request)
-
-        logger.debug("got response from model")
-        yield self.format_chunk({"chunk_type": "message_start"})
-        yield self.format_chunk({"chunk_type": "content_start", "data_type": "text"})
-
-        tool_calls: dict[int, list[Any]] = {}
-
-        async for event in response:
-            # Defensive: skip events with empty or missing choices
-            if not getattr(event, "choices", None):
-                continue
-            choice = event.choices[0]
-
-            if choice.delta.content:
-                yield self.format_chunk(
-                    {"chunk_type": "content_delta", "data_type": "text", "data": choice.delta.content}
-                )
-
-            if hasattr(choice.delta, "reasoning_content") and choice.delta.reasoning_content:
-                yield self.format_chunk(
-                    {
-                        "chunk_type": "content_delta",
-                        "data_type": "reasoning_content",
-                        "data": choice.delta.reasoning_content,
-                    }
-                )
 
-            for tool_call in choice.delta.tool_calls or []:
-                tool_calls.setdefault(tool_call.index, []).append(tool_call)
+        async with openai.AsyncOpenAI(**self.client_args) as client:
+            response = await client.chat.completions.create(**request)
 
-            if choice.finish_reason:
-                break
+            logger.debug("got response from model")
+            yield self.format_chunk({"chunk_type": "message_start"})
+            yield self.format_chunk({"chunk_type": "content_start", "data_type": "text"})
 
-        yield self.format_chunk({"chunk_type": "content_stop", "data_type": "text"})
+            tool_calls: dict[int, list[Any]] = {}
 
-        for tool_deltas in tool_calls.values():
-            yield self.format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_deltas[0]})
+            async for event in response:
+                # Defensive: skip events with empty or missing choices
+                if not getattr(event, "choices", None):
+                    continue
+                choice = event.choices[0]
+
+                if choice.delta.content:
+                    yield self.format_chunk(
+                        {"chunk_type": "content_delta", "data_type": "text", "data": choice.delta.content}
+                    )
+
+                if hasattr(choice.delta, "reasoning_content") and choice.delta.reasoning_content:
+                    yield self.format_chunk(
+                        {
+                            "chunk_type": "content_delta",
+                            "data_type": "reasoning_content",
+                            "data": choice.delta.reasoning_content,
+                        }
+                    )
 
-            for tool_delta in tool_deltas:
-                yield self.format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_delta})
+                for tool_call in choice.delta.tool_calls or []:
+                    tool_calls.setdefault(tool_call.index, []).append(tool_call)
 
-            yield self.format_chunk({"chunk_type": "content_stop", "data_type": "tool"})
+                if choice.finish_reason:
+                    break
 
-        yield self.format_chunk({"chunk_type": "message_stop", "data": choice.finish_reason})
+            yield self.format_chunk({"chunk_type": "content_stop", "data_type": "text"})
 
-        # Skip remaining events as we don't have use for anything except the final usage payload
-        async for event in response:
-            _ = event
+            for tool_deltas in tool_calls.values():
+                yield self.format_chunk({"chunk_type": "content_start", "data_type": "tool", "data": tool_deltas[0]})
 
-        if event.usage:
-            yield self.format_chunk({"chunk_type": "metadata", "data": event.usage})
+                for tool_delta in tool_deltas:
+                    yield self.format_chunk({"chunk_type": "content_delta", "data_type": "tool", "data": tool_delta})
+
+                yield self.format_chunk({"chunk_type": "content_stop", "data_type": "tool"})
+
+            yield self.format_chunk({"chunk_type": "message_stop", "data": choice.finish_reason})
+
+            # Skip remaining events as we don't have use for anything except the final usage payload
+            async for event in response:
+                _ = event
+
+            if event.usage:
+                yield self.format_chunk({"chunk_type": "metadata", "data": event.usage})
 
         logger.debug("finished streaming response from model")
 
@@ -449,11 +449,12 @@ async def structured_output(
         Yields:
             Model events with the last being the structured output.
         """
-        response: ParsedChatCompletion = await self.client.beta.chat.completions.parse(  # type: ignore
-            model=self.get_config()["model_id"],
-            messages=self.format_request(prompt, system_prompt=system_prompt)["messages"],
-            response_format=output_model,
-        )
+        async with openai.AsyncOpenAI(**self.client_args) as client:
+            response: ParsedChatCompletion = await client.beta.chat.completions.parse(
+                model=self.get_config()["model_id"],
+                messages=self.format_request(prompt, system_prompt=system_prompt)["messages"],
+                response_format=output_model,
+            )
 
         parsed: T | None = None
         # Find the first choice with tool_calls

src/strands/telemetry/tracer.py

@@ -207,6 +207,30 @@ def _add_event(self, span: Optional[Span], event_name: str, event_attributes: Di
 
         span.add_event(event_name, attributes=event_attributes)
 
+    def _get_event_name_for_message(self, message: Message) -> str:
+        """Determine the appropriate OpenTelemetry event name for a message.
+
+        According to OpenTelemetry semantic conventions v1.36.0, messages containing tool results
+        should be labeled as 'gen_ai.tool.message' regardless of their role field.
+        This ensures proper categorization of tool responses in traces.
+
+        Note: The GenAI namespace is experimental and may change in future versions.
+
+        Reference: https://github.com/open-telemetry/semantic-conventions/blob/v1.36.0/docs/gen-ai/gen-ai-events.md#event-gen_aitoolmessage
+
+        Args:
+            message: The message to determine the event name for
+
+        Returns:
+            The OpenTelemetry event name (e.g., 'gen_ai.user.message', 'gen_ai.tool.message')
+        """
+        # Check if the message contains a tool result
+        for content_block in message.get("content", []):
+            if "toolResult" in content_block:
+                return "gen_ai.tool.message"
+
+        return f"gen_ai.{message['role']}.message"
+
     def start_model_invoke_span(
         self,
         messages: Messages,
@@ -240,7 +264,7 @@ def start_model_invoke_span(
         for message in messages:
             self._add_event(
                 span,
-                f"gen_ai.{message['role']}.message",
+                self._get_event_name_for_message(message),
                 {"content": serialize(message["content"])},
             )
         return span
@@ -379,7 +403,7 @@ def start_event_loop_cycle_span(
         for message in messages or []:
             self._add_event(
                 span,
-                f"gen_ai.{message['role']}.message",
+                self._get_event_name_for_message(message),
                 {"content": serialize(message["content"])},
             )
 
@@ -456,7 +480,7 @@ def start_agent_span(
         for message in messages:
             self._add_event(
                 span,
-                f"gen_ai.{message['role']}.message",
+                self._get_event_name_for_message(message),
                 {"content": serialize(message["content"])},
             )
 

src/strands/types/_events.py

@@ -169,6 +169,14 @@ def __init__(self, delta: ContentBlockDelta, reasoning_text: str | None) -> None
         super().__init__({"reasoningText": reasoning_text, "delta": delta, "reasoning": True})
 
 
+class ReasoningRedactedContentStreamEvent(ModelStreamEvent):
+    """Event emitted during redacted content streaming."""
+
+    def __init__(self, delta: ContentBlockDelta, redacted_content: bytes | None) -> None:
+        """Initialize with delta and redacted content."""
+        super().__init__({"reasoningRedactedContent": redacted_content, "delta": delta, "reasoning": True})
+
+
 class ReasoningSignatureStreamEvent(ModelStreamEvent):
     """Event emitted during reasoning signature streaming."""
 

tests/fixtures/mocked_model_provider.py

@@ -72,6 +72,10 @@ def map_agent_message_to_events(self, agent_message: Union[Message, RedactionMes
             stop_reason = "guardrail_intervened"
         else:
             for content in agent_message["content"]:
+                if "reasoningContent" in content:
+                    yield {"contentBlockStart": {"start": {}}}
+                    yield {"contentBlockDelta": {"delta": {"reasoningContent": content["reasoningContent"]}}}
+                    yield {"contentBlockStop": {}}
                 if "text" in content:
                     yield {"contentBlockStart": {"start": {}}}
                     yield {"contentBlockDelta": {"delta": {"text": content["text"]}}}

tests/strands/agent/hooks/test_agent_events.py

@@ -387,6 +387,84 @@ async def test_stream_e2e_throttle_and_redact(alist, mock_sleep):
     assert typed_events == []
 
 
+@pytest.mark.asyncio
+async def test_stream_e2e_reasoning_redacted_content(alist):
+    mock_provider = MockedModelProvider(
+        [
+            {
+                "role": "assistant",
+                "content": [
+                    {"reasoningContent": {"redactedContent": b"test_redacted_data"}},
+                    {"text": "Response with redacted reasoning"},
+                ],
+            },
+        ]
+    )
+
+    mock_callback = unittest.mock.Mock()
+    agent = Agent(model=mock_provider, callback_handler=mock_callback)
+
+    stream = agent.stream_async("Test redacted content")
+
+    tru_events = await alist(stream)
+    exp_events = [
+        {"init_event_loop": True},
+        {"start": True},
+        {"start_event_loop": True},
+        {"event": {"messageStart": {"role": "assistant"}}},
+        {"event": {"contentBlockStart": {"start": {}}}},
+        {"event": {"contentBlockDelta": {"delta": {"reasoningContent": {"redactedContent": b"test_redacted_data"}}}}},
+        {
+            **any_props,
+            "reasoningRedactedContent": b"test_redacted_data",
+            "delta": {"reasoningContent": {"redactedContent": b"test_redacted_data"}},
+            "reasoning": True,
+        },
+        {"event": {"contentBlockStop": {}}},
+        {"event": {"contentBlockStart": {"start": {}}}},
+        {"event": {"contentBlockDelta": {"delta": {"text": "Response with redacted reasoning"}}}},
+        {
+            **any_props,
+            "data": "Response with redacted reasoning",
+            "delta": {"text": "Response with redacted reasoning"},
+        },
+        {"event": {"contentBlockStop": {}}},
+        {"event": {"messageStop": {"stopReason": "end_turn"}}},
+        {
+            "message": {
+                "content": [
+                    {"reasoningContent": {"redactedContent": b"test_redacted_data"}},
+                    {"text": "Response with redacted reasoning"},
+                ],
+                "role": "assistant",
+            }
+        },
+        {
+            "result": AgentResult(
+                stop_reason="end_turn",
+                message={
+                    "content": [
+                        {"reasoningContent": {"redactedContent": b"test_redacted_data"}},
+                        {"text": "Response with redacted reasoning"},
+                    ],
+                    "role": "assistant",
+                },
+                metrics=ANY,
+                state={},
+            )
+        },
+    ]
+    assert tru_events == exp_events
+
+    exp_calls = [call(**event) for event in exp_events]
+    act_calls = mock_callback.call_args_list
+    assert act_calls == exp_calls
+
+    # Ensure that all events coming out of the agent are *not* typed events
+    typed_events = [event for event in tru_events if isinstance(event, TypedEvent)]
+    assert typed_events == []
+
+
 @pytest.mark.asyncio
 async def test_event_loop_cycle_text_response_throttling_early_end(
     agenerator,