Python: Decode base64 image, audio and blob content from MCP servers

Author: jaeminbird

python/semantic_kernel/connectors/mcp.py

@@ -111,9 +111,13 @@ def _mcp_content_types_to_kernel_content(
     if isinstance(mcp_type, types.TextContent):
         return [TextContent(text=mcp_type.text, inner_content=mcp_type)]
     if isinstance(mcp_type, types.ImageContent):
-        return [ImageContent(data=mcp_type.data, mime_type=mcp_type.mimeType, inner_content=mcp_type)]
+        return [
+            ImageContent(data=mcp_type.data, data_format="base64", mime_type=mcp_type.mimeType, inner_content=mcp_type)
+        ]
     if isinstance(mcp_type, types.AudioContent):
-        return [AudioContent(data=mcp_type.data, mime_type=mcp_type.mimeType, inner_content=mcp_type)]
+        return [
+            AudioContent(data=mcp_type.data, data_format="base64", mime_type=mcp_type.mimeType, inner_content=mcp_type)
+        ]
     if isinstance(mcp_type, types.ResourceLink):
         return [
             BinaryContent(
@@ -147,6 +151,8 @@ def _mcp_content_types_to_kernel_content(
     return [
         BinaryContent(
             data=mcp_type.resource.blob,
+            data_format="base64",
+            mime_type=mcp_type.resource.mimeType,
             inner_content=mcp_type,
             metadata=mcp_type.annotations.model_dump() if mcp_type.annotations else {},
         )

python/tests/unit/connectors/mcp/test_mcp.py

@@ -1004,3 +1004,39 @@ async def test_mcp_prompt_optional_arguments_are_not_required():
     result = await kernel.invoke(function, code="x = 1")
     assert str(result) == "review this"
     plugin.session.get_prompt.assert_awaited_once_with("review", arguments={"code": "x = 1"})
+
+
+PNG_BYTES = b"\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR"
+PNG_BASE64 = "iVBORw0KGgoAAAANSUhEUg=="
+
+
+@pytest.mark.parametrize(
+    "mcp_content,mime_type",
+    [
+        (types.ImageContent(type="image", data=PNG_BASE64, mimeType="image/png"), "image/png"),
+        (types.AudioContent(type="audio", data=PNG_BASE64, mimeType="audio/wav"), "audio/wav"),
+        (
+            types.EmbeddedResource(
+                type="resource",
+                resource=types.BlobResourceContents(uri="file:///image.png", blob=PNG_BASE64, mimeType="image/png"),
+            ),
+            "image/png",
+        ),
+    ],
+    ids=["image", "audio", "blob_resource"],
+)
+async def test_mcp_base64_content_is_decoded(mcp_content, mime_type):
+    plugin = MCPStdioPlugin(name="test", command="echo")
+    plugin.session = AsyncMock(spec=ClientSession)
+    plugin.session.call_tool.return_value = types.CallToolResult(content=[mcp_content])
+    plugin.session.get_prompt.return_value = types.GetPromptResult(
+        messages=[types.PromptMessage(role="user", content=mcp_content)]
+    )
+
+    [tool_content] = await plugin.call_tool("tool")
+    [prompt_message] = await plugin.get_prompt("prompt")
+
+    for content in (tool_content, prompt_message.items[0]):
+        assert content.data == PNG_BYTES
+        assert content.mime_type == mime_type
+        assert content.data_uri == f"data:{mime_type};base64,{PNG_BASE64}"