feat(mcp): Add EmbeddedResource support to mcp (#726)

---------

Co-authored-by: Dean Schmigelski <[email protected]>

Author: KyMidd

src/strands/tools/mcp/mcp_client.py

@@ -20,8 +20,9 @@
 
 import anyio
 from mcp import ClientSession, ListToolsResult
+from mcp.types import BlobResourceContents, GetPromptResult, ListPromptsResult, TextResourceContents
 from mcp.types import CallToolResult as MCPCallToolResult
-from mcp.types import GetPromptResult, ListPromptsResult
+from mcp.types import EmbeddedResource as MCPEmbeddedResource
 from mcp.types import ImageContent as MCPImageContent
 from mcp.types import TextContent as MCPTextContent
 
@@ -358,8 +359,7 @@ def _handle_tool_result(self, tool_use_id: str, call_tool_result: MCPCallToolRes
         """
         self._log_debug_with_thread("received tool result with %d content items", len(call_tool_result.content))
 
-        # Build a typed list of ToolResultContent. Use a clearer local name to avoid shadowing
-        # and annotate the result for mypy so it knows the intended element type.
+        # Build a typed list of ToolResultContent.
         mapped_contents: list[ToolResultContent] = [
             mc
             for content in call_tool_result.content
@@ -438,7 +438,7 @@ def _background_task(self) -> None:
 
     def _map_mcp_content_to_tool_result_content(
         self,
-        content: MCPTextContent | MCPImageContent | Any,
+        content: MCPTextContent | MCPImageContent | MCPEmbeddedResource | Any,
     ) -> Union[ToolResultContent, None]:
         """Maps MCP content types to tool result content types.
 
@@ -462,6 +462,58 @@ def _map_mcp_content_to_tool_result_content(
                     "source": {"bytes": base64.b64decode(content.data)},
                 }
             }
+        elif isinstance(content, MCPEmbeddedResource):
+            """
+            TODO: Include URI information in results.
+                Models may find it useful to be aware not only of the information,
+                but the location of the information too.
+
+                This may be difficult without taking an opinionated position. For example,
+                a content block may need to indicate that the following Image content block
+                is of particular URI.
+            """
+
+            self._log_debug_with_thread("mapping MCP embedded resource content")
+
+            resource = content.resource
+            if isinstance(resource, TextResourceContents):
+                return {"text": resource.text}
+            elif isinstance(resource, BlobResourceContents):
+                try:
+                    raw_bytes = base64.b64decode(resource.blob)
+                except Exception:
+                    self._log_debug_with_thread("embedded resource blob could not be decoded - dropping")
+                    return None
+
+                if resource.mimeType and (
+                    resource.mimeType.startswith("text/")
+                    or resource.mimeType
+                    in (
+                        "application/json",
+                        "application/xml",
+                        "application/javascript",
+                        "application/yaml",
+                        "application/x-yaml",
+                    )
+                    or resource.mimeType.endswith(("+json", "+xml"))
+                ):
+                    try:
+                        return {"text": raw_bytes.decode("utf-8", errors="replace")}
+                    except Exception:
+                        pass
+
+                if resource.mimeType in MIME_TO_FORMAT:
+                    return {
+                        "image": {
+                            "format": MIME_TO_FORMAT[resource.mimeType],
+                            "source": {"bytes": raw_bytes},
+                        }
+                    }
+
+                self._log_debug_with_thread("embedded resource blob with non-textual/unknown mimeType - dropping")
+                return None
+
+            return None  # type: ignore[unreachable]  # Defensive: future MCP resource types
         else:
             self._log_debug_with_thread("unhandled content type: %s - dropping content", content.__class__.__name__)
             return None

tests/strands/tools/mcp/test_mcp_client.py

@@ -1,3 +1,4 @@
+import base64
 import time
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -541,3 +542,149 @@ def slow_transport():
     assert client._background_thread_session is None
     assert client._background_thread_event_loop is None
     assert not client._init_future.done()  # New future created
+
+
+def test_call_tool_sync_embedded_nested_text(mock_transport, mock_session):
+    """EmbeddedResource.resource (uri + text) should map to plain text content."""
+    embedded_resource = {
+        "type": "resource",  # required literal
+        "resource": {
+            "uri": "mcp://resource/embedded-text-1",
+            "text": "inner text",
+            "mimeType": "text/plain",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-text", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert result["content"][0]["text"] == "inner text"
+
+
+def test_call_tool_sync_embedded_nested_base64_textual_mime(mock_transport, mock_session):
+    """EmbeddedResource.resource (uri + blob with textual MIME) should decode to text."""
+
+    payload = base64.b64encode(b'{"k":"v"}').decode()
+
+    embedded_resource = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-blob-1",
+            # NOTE: blob is a STRING, mimeType is sibling
+            "blob": payload,
+            "mimeType": "application/json",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-blob", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert result["content"][0]["text"] == '{"k":"v"}'
+
+
+def test_call_tool_sync_embedded_image_blob(mock_transport, mock_session):
+    """EmbeddedResource.resource (blob with image MIME) should map to image content."""
+    # Read yellow.png file
+    with open("tests_integ/yellow.png", "rb") as image_file:
+        png_data = image_file.read()
+    payload = base64.b64encode(png_data).decode()
+
+    embedded_resource = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-image",
+            "blob": payload,
+            "mimeType": "image/png",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-image", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert "image" in result["content"][0]
+        assert result["content"][0]["image"]["format"] == "png"
+        assert "bytes" in result["content"][0]["image"]["source"]
+
+
+def test_call_tool_sync_embedded_non_textual_blob_dropped(mock_transport, mock_session):
+    """EmbeddedResource.resource (blob with non-textual/unknown MIME) should be dropped."""
+    payload = base64.b64encode(b"\x00\x01\x02\x03").decode()
+
+    embedded_resource = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-binary",
+            "blob": payload,
+            "mimeType": "application/octet-stream",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-binary", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 0  # Content should be dropped
+
+
+def test_call_tool_sync_embedded_multiple_textual_mimes(mock_transport, mock_session):
+    """EmbeddedResource with different textual MIME types should decode to text."""
+
+    # Test YAML content
+    yaml_content = base64.b64encode(b"key: value\nlist:\n  - item1\n  - item2").decode()
+    embedded_resource = {
+        "type": "resource",
+        "resource": {
+            "uri": "mcp://resource/embedded-yaml",
+            "blob": yaml_content,
+            "mimeType": "application/yaml",
+        },
+    }
+    mock_session.call_tool.return_value = MCPCallToolResult(isError=False, content=[embedded_resource])
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-yaml", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 1
+        assert "key: value" in result["content"][0]["text"]
+
+
+def test_call_tool_sync_embedded_unknown_resource_type_dropped(mock_transport, mock_session):
+    """EmbeddedResource with unknown resource type should be dropped for forward compatibility."""
+
+    # Mock an unknown resource type that's neither TextResourceContents nor BlobResourceContents
+    class UnknownResourceContents:
+        def __init__(self):
+            self.uri = "mcp://resource/unknown-type"
+            self.mimeType = "application/unknown"
+            self.data = "some unknown data"
+
+    # Create a mock embedded resource with unknown resource type
+    mock_embedded_resource = MagicMock()
+    mock_embedded_resource.resource = UnknownResourceContents()
+
+    mock_session.call_tool.return_value = MagicMock(
+        isError=False, content=[mock_embedded_resource], structuredContent=None
+    )
+
+    with MCPClient(mock_transport["transport_callable"]) as client:
+        result = client.call_tool_sync(tool_use_id="er-unknown", name="get_file_contents", arguments={})
+
+        mock_session.call_tool.assert_called_once_with("get_file_contents", {}, None)
+        assert result["status"] == "success"
+        assert len(result["content"]) == 0  # Unknown resource type should be dropped

tests_integ/mcp/echo_server.py

@@ -15,7 +15,11 @@
     $ python echo_server.py
 """
 
+import base64
+from typing import Literal
+
 from mcp.server import FastMCP
+from mcp.types import BlobResourceContents, EmbeddedResource, TextResourceContents
 from pydantic import BaseModel
 
 
@@ -46,6 +50,48 @@ def echo(to_echo: str) -> str:
     def echo_with_structured_content(to_echo: str) -> EchoResponse:
         return EchoResponse(echoed=to_echo, message_length=len(to_echo))
 
+    @mcp.tool(description="Get current weather information for a location")
+    def get_weather(location: Literal["New York", "London", "Tokyo"] = "New York"):
+        """Get weather data including forecasts and alerts for the specified location"""
+        if location.lower() == "new york":
+            return [
+                EmbeddedResource(
+                    type="resource",
+                    resource=TextResourceContents(
+                        uri="https://weather.api/forecast/nyc",
+                        mimeType="text/plain",
+                        text="Current weather in New York: 72°F, partly cloudy with light winds.",
+                    ),
+                )
+            ]
+        elif location.lower() == "london":
+            return [
+                EmbeddedResource(
+                    type="resource",
+                    resource=BlobResourceContents(
+                        uri="https://weather.api/data/london.json",
+                        mimeType="application/json",
+                        blob=base64.b64encode(
+                            '{"temperature": 18, "condition": "rainy", "humidity": 85}'.encode()
+                        ).decode(),
+                    ),
+                )
+            ]
+        elif location.lower() == "tokyo":
+            # Read yellow.png file for weather icon
+            with open("tests_integ/yellow.png", "rb") as image_file:
+                png_data = image_file.read()
+            return [
+                EmbeddedResource(
+                    type="resource",
+                    resource=BlobResourceContents(
+                        uri="https://weather.api/icons/sunny.png",
+                        mimeType="image/png",
+                        blob=base64.b64encode(png_data).decode(),
+                    ),
+                )
+            ]
+
     mcp.run(transport="stdio")