mirror of
https://github.com/mcp-use/mcp-use.git
synced 2026-09-14 18:06:01 +08:00
Pietro Zullo
2ddad74dab
* fix(python): convert MCP tool results to LangChain content blocks
Previously, tool results were converted using `str(tool_result.content)`,
which produced Python repr strings like
`[TextContent(type='text', text='...')]` instead of the actual content.
Now properly maps MCP content types to LangChain format:
- TextContent → plain string or {"type": "text", ...}
- ImageContent → {"type": "image", "source_type": "base64", ...}
- AudioContent → {"type": "audio", "source_type": "base64", ...}
- EmbeddedResource → text or file block depending on resource type
Single TextContent results (the most common case) are returned as
plain strings for backward compatibility.
* test(python): cover langchain tool content conversion
* style(python): format langchain adapter
* test(python): cover langchain block tool messages
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""Unit tests for LangChain adapter content conversion."""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from mcp.types import (
|
|
AudioContent,
|
|
BlobResourceContents,
|
|
CallToolResult,
|
|
EmbeddedResource,
|
|
ImageContent,
|
|
TextContent,
|
|
TextResourceContents,
|
|
Tool,
|
|
)
|
|
|
|
from mcp_use.agents.adapters.langchain_adapter import LangChainAdapter, _mcp_content_to_langchain
|
|
|
|
|
|
class TestLangChainAdapterContentConversion:
|
|
"""Tests for MCP-to-LangChain content conversion."""
|
|
|
|
def test_single_text_content_returns_plain_string(self):
|
|
"""Single text results should preserve the simple string contract."""
|
|
result = _mcp_content_to_langchain([TextContent(type="text", text="hello world")])
|
|
|
|
assert result == "hello world"
|
|
|
|
def test_mixed_content_returns_langchain_blocks(self):
|
|
"""Mixed content should map to LangChain block dictionaries."""
|
|
result = _mcp_content_to_langchain(
|
|
[
|
|
TextContent(type="text", text="intro"),
|
|
ImageContent(type="image", data="aGVsbG8=", mimeType="image/png"),
|
|
AudioContent(type="audio", data="d29ybGQ=", mimeType="audio/mpeg"),
|
|
EmbeddedResource(
|
|
type="resource",
|
|
resource=TextResourceContents(uri="file:///tmp/readme.txt", text="resource text"),
|
|
),
|
|
EmbeddedResource(
|
|
type="resource",
|
|
resource=BlobResourceContents(uri="file:///tmp/data.bin", blob="ZGF0YQ=="),
|
|
),
|
|
]
|
|
)
|
|
|
|
assert result == [
|
|
{"type": "text", "text": "intro"},
|
|
{
|
|
"type": "image",
|
|
"source_type": "base64",
|
|
"data": "aGVsbG8=",
|
|
"mime_type": "image/png",
|
|
},
|
|
{
|
|
"type": "audio",
|
|
"source_type": "base64",
|
|
"data": "d29ybGQ=",
|
|
"mime_type": "audio/mpeg",
|
|
},
|
|
{"type": "text", "text": "resource text"},
|
|
{
|
|
"type": "file",
|
|
"source_type": "base64",
|
|
"data": "ZGF0YQ==",
|
|
"mime_type": "application/octet-stream",
|
|
},
|
|
]
|
|
|
|
def test_unknown_embedded_resource_falls_back_to_text(self):
|
|
"""Unexpected embedded resource payloads should not be dropped silently."""
|
|
result = _mcp_content_to_langchain(
|
|
[
|
|
EmbeddedResource.model_construct(
|
|
type="resource",
|
|
resource={"unexpected": "value"},
|
|
)
|
|
]
|
|
)
|
|
|
|
assert result == "{'unexpected': 'value'}"
|
|
|
|
|
|
class TestLangChainAdapterToolExecution:
|
|
"""Tests for LangChain tool execution behavior."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_error_call_tool_result_returns_formatted_error(self):
|
|
"""MCP error results should be surfaced as structured LangChain tool errors."""
|
|
adapter = LangChainAdapter()
|
|
connector = MagicMock()
|
|
connector.call_tool = AsyncMock(
|
|
return_value=CallToolResult(
|
|
content=[TextContent(type="text", text="tool failed")],
|
|
isError=True,
|
|
)
|
|
)
|
|
|
|
tool = Tool(
|
|
name="failing_tool",
|
|
description="A tool that fails",
|
|
inputSchema={"type": "object", "properties": {}},
|
|
)
|
|
langchain_tool = adapter._convert_tool(tool, connector)
|
|
|
|
assert langchain_tool is not None
|
|
|
|
result = await langchain_tool._arun()
|
|
|
|
connector.call_tool.assert_awaited_once_with("failing_tool", {})
|
|
assert result["error"] == "RuntimeError"
|
|
assert result["details"] == "tool failed"
|
|
assert result["tool"] == "failing_tool"
|
|
assert result["tool_content"] == "tool failed"
|