fix(langchain): name the content type MCP conversion could not handle

`ContentBlock` is closed at type-check time, but only as closed as the
installed `mcp`. A caller on a newer SDK could reach the fallback, where
`assert_never` raised an `AssertionError` that reads as "impossible" rather
than "your `mcp` is newer than this adapter".

Both fallbacks now raise a `ValueError` naming the type that arrived, and the
embedded-resource branch checks for `BlobResourceContents` instead of assuming
anything that is not text carries a `blob` — which would have been an
`AttributeError` on a resource kind the SDK adds later.

Exhaustiveness is still enforced statically: mypy proves both lines
unreachable, so the `unreachable` ignores become unused-ignore errors under
`strict` the moment either union grows a member.
This commit is contained in:
Sydney Runkle committed 2026-08-27 15:13:47 -07:00
1 parent 65773d2ae2
commit bd2f725f78
2 files changed
+54 -9

No files matched your search

+21 -8
View File
@@ -20,6 +20,7 @@ from langchain_core.messages.content import (
from langchain_core.tools import BaseTool, StructuredTool, ToolException
from mcp.types import (
AudioContent,
BlobResourceContents,
ContentBlock,
EmbeddedResource,
ImageContent,
@@ -27,7 +28,6 @@ from mcp.types import (
TextContent,
TextResourceContents,
)
from typing_extensions import assert_never
from langchain.mcp.elicitation import _call_tool_with_interrupts
@@ -139,14 +139,27 @@ def _convert_content_block(content: ContentBlock) -> ToolMessageContentBlock:
resource = content.resource
if isinstance(resource, TextResourceContents):
return create_text_block(text=resource.text)
mime_type = resource.mime_type or None
if mime_type and mime_type.startswith("image/"):
return create_image_block(base64=resource.blob, mime_type=mime_type)
return create_file_block(base64=resource.blob, mime_type=mime_type)
if isinstance(resource, BlobResourceContents):
mime_type = resource.mime_type or None
if mime_type and mime_type.startswith("image/"):
return create_image_block(base64=resource.blob, mime_type=mime_type)
return create_file_block(base64=resource.blob, mime_type=mime_type)
# Unreachable while the SDK's resource union holds; see the note below.
msg = f"Unknown embedded resource type: {type(resource).__name__}" # type: ignore[unreachable]
raise ValueError(msg)
# `ContentBlock` is a closed union, so a block reaching here means the MCP
# SDK grew a content type this conversion has not been taught yet.
assert_never(content)
# Both unions are closed at type-check time, so mypy proves these two lines
# unreachable — and the `unreachable` ignores become unused-ignore errors the
# moment the SDK grows a member, which is what keeps this exhaustive.
#
# The runtime guards still earn their place: a union is only as closed as the
# installed `mcp`, and a caller on a newer SDK is better served by a named
# type than by the bare `AssertionError` an `assert_never` would raise.
msg = ( # type: ignore[unreachable]
f"Unknown MCP content type: {type(content).__name__}. This usually means "
"the installed `mcp` is newer than the version this adapter supports."
)
raise ValueError(msg)
def _convert_call_tool_result(
@@ -2,7 +2,7 @@
from __future__ import annotations
from typing import Any, Literal
from typing import Any, Literal, cast
from unittest.mock import AsyncMock
import pytest
@@ -25,6 +25,14 @@ from langchain.mcp import convert_mcp_tool_to_langchain_tool
from langchain.mcp.tools import _convert_call_tool_result, _convert_content_block
class _VideoContent:
"""Stand-in for an MCP content type newer than this adapter."""
class _VideoResource:
"""Stand-in for an embedded resource kind newer than this adapter."""
def _blocks_without_ids(content: Any) -> list[dict[str, Any]]:
"""Drop the generated block ids so content can be compared literally."""
return [{key: value for key, value in block.items() if key != "id"} for block in content]
@@ -245,3 +253,27 @@ def test_embedded_blob_resource_type_follows_its_mime_type(
def test_audio_content_is_not_yet_supported() -> None:
with pytest.raises(NotImplementedError, match="audio"):
_convert_content_block(AudioContent(type="audio", data="AAAA", mimeType="audio/wav"))
def test_an_unknown_content_type_names_itself_rather_than_asserting() -> None:
"""A content type the conversion has not been taught names itself.
`ContentBlock` is closed at type-check time, but only as closed at runtime as
the installed `mcp`. A caller on a newer SDK should learn which type arrived,
not catch a bare `AssertionError`.
"""
with pytest.raises(ValueError, match="Unknown MCP content type: _VideoContent"):
_convert_content_block(cast("Any", _VideoContent()))
def test_an_unknown_embedded_resource_type_names_itself() -> None:
"""An embedded resource that is neither text nor blob raises rather than guessing."""
embedded = EmbeddedResource(
type="resource",
resource=TextResourceContents(uri="file:///notes.txt", text="notes"),
)
# Bypass validation to stand in for a resource kind the SDK might add later.
object.__setattr__(embedded, "resource", cast("Any", _VideoResource()))
with pytest.raises(ValueError, match="Unknown embedded resource type: _VideoResource"):
_convert_content_block(embedded)