Add EmbeddedResource support to mcp

src/strands/tools/mcp/mcp_client.py

@@ -23,6 +23,7 @@
 from mcp.types import GetPromptResult, ListPromptsResult
 from mcp.types import ImageContent as MCPImageContent
 from mcp.types import TextContent as MCPTextContent
+from mcp.types import EmbeddedResource as MCPEmbeddedResource  # <-- minimal add
 
 from ...types import PaginatedList
 from ...types.exceptions import MCPClientInitializationError
@@ -410,6 +411,88 @@ def _map_mcp_content_to_tool_result_content(
                     "source": {"bytes": base64.b64decode(content.data)},
                 }
             }
+        # If EmbeddedResource
+        elif isinstance(content, MCPEmbeddedResource):
+            self._log_debug_with_thread("mapping MCP embedded resource content")
+            res = getattr(content, "resource", None)
+            if res is None:
+                self._log_debug_with_thread("embedded resource has no 'resource' field - dropping")
+                return None
+
+            # Support both pydantic model and dict access
+            def _get(attr: str) -> Any:
+                if hasattr(res, attr):
+                    return getattr(res, attr)
+                if isinstance(res, dict):
+                    return res.get(attr)
+                return None
+
+            text_val = _get("text")
+            if text_val:
+                return {"text": text_val}
+
+            blob_val = _get("blob")
+            mime_type = _get("mimeType")
+
+            if blob_val is not None:
+                # blob is a base64 string in current mcp schema
+                raw_bytes: Optional[bytes]
+                try:
+                    if isinstance(blob_val, (bytes, bytearray)):
+                        raw_bytes = bytes(blob_val)
+                    elif isinstance(blob_val, str):
+                        raw_bytes = base64.b64decode(blob_val)
+                    else:
+                        raw_bytes = None
+                except Exception:
+                    raw_bytes = None
+
+                if raw_bytes is None:
+                    self._log_debug_with_thread("embedded resource blob could not be decoded - dropping")
+                    return None
+
+                def _is_textual(mt: Optional[str]) -> bool:
+                    if not mt:
+                        return False
+                    if mt.startswith("text/"):
+                        return True
+                    textual = (
+                        "application/json",
+                        "application/xml",
+                        "application/javascript",
+                        "application/x-yaml",
+                        "application/yaml",
+                        "application/xhtml+xml",
+                    )
+                    if mt in textual or mt.endswith("+json") or mt.endswith("+xml"):
+                        return True
+                    return False
+
+                if _is_textual(mime_type):
+                    try:
+                        return {"text": raw_bytes.decode("utf-8", errors="replace")}
+                    except Exception:
+                        pass
+
+                if mime_type in MIME_TO_FORMAT:
+                    return {
+                        "image": {
+                            "format": MIME_TO_FORMAT[mime_type],
+                            "source": {"bytes": raw_bytes},
+                        }
+                    }
+
+                self._log_debug_with_thread("embedded resource blob with non-textual/unknown mimeType - dropping")
+                return None
+
+            # Handle URI-only resources
+            uri = _get("uri")
+            if uri:
+                return {"text": f"[embedded resource] {uri} ({mime_type or 'unknown mime'})"}
+
+            # Make sure we return in all paths
+            self._log_debug_with_thread("embedded resource had no usable text/blob/uri; dropping")
+            return None
         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,4 +1,5 @@
 import time
+import base64
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
@@ -466,3 +467,52 @@ def test_get_prompt_sync_session_not_active():
 
     with pytest.raises(MCPClientInitializationError, match="client session is not running"):
         client.get_prompt_sync("test_prompt_id", {})
+
+
+def test_call_tool_sync_embedded_nested_text(mock_transport, mock_session):
+    """EmbeddedResource.resource (uri + text) should map to plain text content."""
+    er = {
+        "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=[er])
+
+    from strands.tools.mcp import MCPClient
+    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."""
+    import base64
+    from mcp.types import CallToolResult as MCPCallToolResult
+    payload = base64.b64encode(b'{"k":"v"}').decode()
+
+    er = {
+        "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=[er])
+
+    from strands.tools.mcp import MCPClient
+    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"}'