feat(langchain): add langchain.mcp for loading MCP tools

Adds an optional `langchain.mcp` module that turns the tools on an MCP server into
LangChain tools. Connection concerns (transport, multi-server configuration, auth,
protocol negotiation) are handled by the FastMCP client; this package only does the
conversion.

`load_mcp_tools(client)` returns `list[BaseTool]`. Install with `langchain[mcp]`, which
pulls the lightweight `fastmcp-slim[client]`.

libs/langchain_v1/langchain/mcp/__init__.py

@@ -0,0 +1,38 @@
+"""LangChain integration for the Model Context Protocol (MCP).
+
+Convert the tools exposed by an MCP server into native LangChain tools. Connections are
+managed by [FastMCP](https://gofastmcp.com); this package only converts tools. Requires
+the optional `mcp` dependency:
+
+```bash
+pip install "langchain[mcp]"
+```
+
+Example:
+    ```python
+    from fastmcp import Client
+
+    from langchain.mcp import load_mcp_tools
+
+    async with Client("https://example.com/mcp") as client:
+        tools = await load_mcp_tools(client)
+    ```
+"""
+
+from __future__ import annotations
+
+try:
+    import fastmcp  # noqa: F401
+except ImportError as e:  # pragma: no cover
+    msg = (
+        "langchain.mcp requires the optional 'mcp' dependency. "
+        'Install it with: pip install "langchain[mcp]"'
+    )
+    raise ModuleNotFoundError(msg) from e
+
+from langchain.mcp.tools import convert_mcp_tool, load_mcp_tools
+
+__all__ = [
+    "convert_mcp_tool",
+    "load_mcp_tools",
+]

libs/langchain_v1/langchain/mcp/tools.py

@@ -0,0 +1,154 @@
+"""Load tools from an MCP server (via a FastMCP client) as LangChain tools."""
+
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Any
+
+from langchain_core.messages.content import (
+    ContentBlock,
+    create_file_block,
+    create_image_block,
+    create_text_block,
+)
+from langchain_core.tools import StructuredTool, ToolException
+from mcp.types import (
+    AudioContent,
+    ImageContent,
+    ResourceLink,
+    TextContent,
+    TextResourceContents,
+)
+
+if TYPE_CHECKING:
+    from fastmcp import Client
+    from fastmcp.client.client import CallToolResult
+    from fastmcp.client.transports import ClientTransport
+    from langchain_core.tools import BaseTool
+    from mcp.types import ContentBlock as MCPContentBlock
+    from mcp.types import Tool as MCPTool
+
+# Result artifact carrying the server's structured (JSON) output, when provided.
+_STRUCTURED_CONTENT_KEY = "structured_content"
+
+
+async def load_mcp_tools(client: Client[ClientTransport]) -> list[BaseTool]:
+    """Load the tools exposed by an MCP server as LangChain tools.
+
+    Connection, transport, multi-server configuration, authentication, and protocol
+    negotiation are all handled by the FastMCP client; this function only converts the
+    server's tools into LangChain tools.
+
+    Args:
+        client: A FastMCP [`Client`][fastmcp.Client]. Enter it as an async context
+            manager (`async with client: ...`) to reuse a single connection across tool
+            calls; otherwise FastMCP connects per call as needed.
+
+    Returns:
+        The server's tools as LangChain tools.
+    """
+    tools = await client.list_tools()
+    return [convert_mcp_tool(tool, client) for tool in tools]
+
+
+def convert_mcp_tool(tool: MCPTool, client: Client[ClientTransport]) -> BaseTool:
+    """Convert a single MCP tool definition into a LangChain tool.
+
+    Args:
+        tool: The MCP tool definition to convert.
+        client: FastMCP client used to invoke the tool.
+
+    Returns:
+        A [`StructuredTool`][langchain_core.tools.StructuredTool] wrapping the MCP tool.
+    """
+    tool_name = tool.name
+
+    async def call_tool(**arguments: Any) -> tuple[list[ContentBlock], dict[str, Any] | None]:
+        result = await client.call_tool(tool_name, arguments, raise_on_error=False)
+        return _convert_call_result(result)
+
+    return StructuredTool(
+        name=tool_name,
+        description=tool.description or "",
+        args_schema=tool.input_schema,
+        coroutine=call_tool,
+        response_format="content_and_artifact",
+        metadata=_tool_metadata(tool),
+    )
+
+
+def _tool_metadata(tool: MCPTool) -> dict[str, Any] | None:
+    """Collect MCP-specific metadata (annotations, `_meta`) for the LangChain tool."""
+    metadata: dict[str, Any] = {}
+    if tool.annotations is not None:
+        metadata["annotations"] = tool.annotations.model_dump(exclude_none=True)
+    if tool.meta is not None:
+        metadata["_meta"] = tool.meta
+    return metadata or None
+
+
+def _convert_call_result(
+    result: CallToolResult,
+) -> tuple[list[ContentBlock], dict[str, Any] | None]:
+    """Convert an MCP tool-call result into LangChain content and an artifact.
+
+    Args:
+        result: The result returned by the FastMCP client.
+
+    Returns:
+        A `(content, artifact)` tuple. `content` is a list of content blocks; `artifact`
+        carries the server's `structuredContent`, if any.
+
+    Raises:
+        ToolException: If the server reported the tool call as an error.
+    """
+    if result.is_error:
+        text = "\n".join(b.text for b in result.content if isinstance(b, TextContent))
+        raise ToolException(text or "MCP tool call failed.")
+
+    artifact: dict[str, Any] | None = None
+    if result.structured_content is not None:
+        artifact = {_STRUCTURED_CONTENT_KEY: result.structured_content}
+
+    blocks = [_convert_content_block(block) for block in result.content]
+    if not blocks:
+        # Surface structured-only results so the model still sees a payload.
+        fallback = json.dumps(result.structured_content) if result.structured_content else ""
+        blocks = [create_text_block(fallback)]
+
+    return blocks, artifact
+
+
+def _convert_content_block(block: MCPContentBlock) -> ContentBlock:
+    """Map an MCP content block to the equivalent LangChain content block.
+
+    Text maps to a text block and images to an image block. Everything else — audio,
+    binary blobs, resource links, and embedded non-text resources — maps to a file block
+    (text resources map to a text block). No provider currently accepts audio or video in
+    a tool result, so they are represented as file attachments rather than dedicated
+    audio/video blocks.
+    """
+    if isinstance(block, TextContent):
+        return create_text_block(block.text)
+    if isinstance(block, ImageContent):
+        return create_image_block(base64=block.data, mime_type=block.mime_type)
+    if isinstance(block, AudioContent):
+        return create_file_block(base64=block.data, mime_type=block.mime_type)
+    if isinstance(block, ResourceLink):
+        return _resource_block(mime_type=block.mime_type, url=str(block.uri))
+    resource = block.resource
+    if isinstance(resource, TextResourceContents):
+        return create_text_block(resource.text)
+    return _resource_block(mime_type=resource.mime_type, base64=resource.blob)
+
+
+def _resource_block(
+    *,
+    mime_type: str | None,
+    url: str | None = None,
+    base64: str | None = None,
+) -> ContentBlock:
+    """Route a linked or binary resource to an image block or a file block by MIME type."""
+    if (mime_type or "").split("/", 1)[0] == "image":
+        return create_image_block(url=url, base64=base64, mime_type=mime_type)
+    return create_file_block(url=url, base64=base64, mime_type=mime_type)

libs/langchain_v1/pyproject.toml

@@ -49,6 +49,7 @@ deepseek = ["langchain-deepseek"]
 xai = ["langchain-xai"]
 perplexity = ["langchain-perplexity"]
 meta = ["langchain-meta"]
+mcp = ["fastmcp-slim[client]>=4.0.0b1"]
 
 [project.urls]
 Homepage = "https://docs.langchain.com/"
@@ -75,6 +76,7 @@ test = [
     "blockbuster>=1.5.26,<1.6.0",
     "langchain-tests>=1.1.9,<2.0.0",
     "langchain-openai",
+    "fastmcp>=4.0.0b1",
 ]
 lint = [
     "ruff>=0.15.0,<0.17.0",

libs/langchain_v1/tests/integration_tests/mcp/_server.py

@@ -0,0 +1,21 @@
+"""Minimal MCP server used by integration tests (run over stdio)."""
+
+from fastmcp import FastMCP
+
+server = FastMCP("langchain-mcp-test")
+
+
+@server.tool
+def add(a: int, b: int) -> int:
+    """Add two integers."""
+    return a + b
+
+
+@server.tool
+def greet(name: str) -> str:
+    """Greet someone by name."""
+    return f"Hello, {name}!"
+
+
+if __name__ == "__main__":
+    server.run()

libs/langchain_v1/tests/integration_tests/mcp/test_client.py

@@ -0,0 +1,32 @@
+"""Integration tests for `langchain.mcp` against a real stdio MCP server."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from fastmcp import Client
+from fastmcp.client.transports import StdioTransport
+
+from langchain.mcp import load_mcp_tools
+
+_SERVER_SCRIPT = str(Path(__file__).parent / "_server.py")
+
+_ADD_CALL = {"type": "tool_call", "name": "add", "args": {"a": 2, "b": 3}, "id": "1"}
+
+
+def _client() -> Client[StdioTransport]:
+    return Client(StdioTransport(command=sys.executable, args=[_SERVER_SCRIPT]))
+
+
+async def test_load_mcp_tools_lists_server_tools() -> None:
+    async with _client() as client:
+        tools = await load_mcp_tools(client)
+    assert {"add", "greet"} <= {tool.name for tool in tools}
+
+
+async def test_call_tool_end_to_end() -> None:
+    async with _client() as client:
+        tools = {tool.name: tool for tool in await load_mcp_tools(client)}
+        message = await tools["add"].ainvoke(_ADD_CALL)
+    assert "5" in str(message.content)

libs/langchain_v1/tests/unit_tests/mcp/test_tools.py

@@ -0,0 +1,164 @@
+"""Unit tests for `langchain.mcp` tool conversion.
+
+These exercise the pure conversion logic with a stubbed client; the live transport path
+is covered by the integration tests.
+"""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+from typing import Any, get_type_hints
+
+import pytest
+from langchain_core.tools import StructuredTool, ToolException
+from mcp.types import (
+    AudioContent,
+    BlobResourceContents,
+    EmbeddedResource,
+    ImageContent,
+    ResourceLink,
+    TextContent,
+    TextResourceContents,
+    Tool,
+)
+
+from langchain.mcp import convert_mcp_tool, load_mcp_tools
+from langchain.mcp.tools import _convert_call_result, _convert_content_block
+
+_ADD_CALL = {"type": "tool_call", "name": "add", "args": {"a": 2, "b": 3}, "id": "1"}
+
+_ADD_TOOL = Tool.model_validate(
+    {
+        "name": "add",
+        "description": "Add two integers.",
+        "inputSchema": {
+            "type": "object",
+            "properties": {"a": {"type": "integer"}, "b": {"type": "integer"}},
+            "required": ["a", "b"],
+        },
+    }
+)
+
+
+def _result(
+    *,
+    content: list[Any],
+    structured: dict[str, Any] | None = None,
+    is_error: bool = False,
+) -> Any:
+    """Build a minimal stand-in for a FastMCP `CallToolResult`."""
+    return SimpleNamespace(content=content, structured_content=structured, is_error=is_error)
+
+
+class _StubClient:
+    """Minimal stand-in for a FastMCP `Client`."""
+
+    async def list_tools(self) -> list[Tool]:
+        return [_ADD_TOOL]
+
+    async def call_tool(self, name: str, arguments: dict[str, Any], *, raise_on_error: bool) -> Any:  # noqa: ARG002
+        return _result(content=[TextContent(type="text", text="5")], structured={"result": 5})
+
+
+def test_convert_content_block_text() -> None:
+    block = _convert_content_block(TextContent(type="text", text="hi"))
+    assert block["type"] == "text"
+    assert block["text"] == "hi"
+
+
+def test_convert_content_block_image_and_audio() -> None:
+    image = _convert_content_block(ImageContent(type="image", data="AAA", mime_type="image/png"))
+    assert image["type"] == "image"
+    assert image["base64"] == "AAA"
+    assert image["mime_type"] == "image/png"
+
+    # Audio has no dedicated tool-result block across providers, so it maps to a file.
+    audio = _convert_content_block(AudioContent(type="audio", data="BBB", mime_type="audio/wav"))
+    assert audio["type"] == "file"
+    assert audio["base64"] == "BBB"
+    assert audio["mime_type"] == "audio/wav"
+
+
+def test_convert_resource_link_routes_by_mime_type() -> None:
+    image = _convert_content_block(
+        ResourceLink(
+            type="resource_link", name="i", uri="https://ex.com/i.png", mime_type="image/png"
+        )
+    )
+    assert image["type"] == "image"
+    assert "i.png" in image["url"]
+
+    doc = _convert_content_block(
+        ResourceLink(
+            type="resource_link", name="d", uri="https://ex.com/d.pdf", mime_type="application/pdf"
+        )
+    )
+    assert doc["type"] == "file"
+    assert "d.pdf" in doc["url"]
+
+
+def test_convert_embedded_resource() -> None:
+    text = _convert_content_block(
+        EmbeddedResource(
+            type="resource",
+            resource=TextResourceContents(
+                uri="file:///a.txt", text="hello", mime_type="text/plain"
+            ),
+        )
+    )
+    assert text["type"] == "text"
+    assert text["text"] == "hello"
+
+    blob = _convert_content_block(
+        EmbeddedResource(
+            type="resource",
+            resource=BlobResourceContents(uri="file:///a.png", blob="AAA", mime_type="image/png"),
+        )
+    )
+    assert blob["type"] == "image"
+    assert blob["base64"] == "AAA"
+
+
+def test_convert_call_result_text_and_artifact() -> None:
+    content, artifact = _convert_call_result(
+        _result(content=[TextContent(type="text", text="5")], structured={"result": 5})
+    )
+    assert content[0]["type"] == "text"
+    assert content[0]["text"] == "5"
+    assert artifact == {"structured_content": {"result": 5}}
+
+
+def test_convert_call_result_error_raises() -> None:
+    result = _result(content=[TextContent(type="text", text="boom")], is_error=True)
+    with pytest.raises(ToolException, match="boom"):
+        _convert_call_result(result)
+
+
+async def test_load_mcp_tools_and_invoke() -> None:
+    client: Any = _StubClient()
+    tools = await load_mcp_tools(client)
+    assert [tool.name for tool in tools] == ["add"]
+    assert tools[0].description == "Add two integers."
+
+    message = await tools[0].ainvoke(_ADD_CALL)
+    assert "5" in str(message.content)
+
+
+async def test_convert_mcp_tool_invocation() -> None:
+    client: Any = _StubClient()
+    tool = convert_mcp_tool(_ADD_TOOL, client)
+    message = await tool.ainvoke(_ADD_CALL)
+    assert "5" in str(message.content)
+
+
+def test_tool_annotations_resolve_at_runtime() -> None:
+    """`create_agent` introspects the tool coroutine with `get_type_hints`.
+
+    Any annotation it references (e.g. `ContentBlock`) must be importable at runtime, not
+    only under `TYPE_CHECKING`.
+    """
+    client: Any = _StubClient()
+    tool = convert_mcp_tool(_ADD_TOOL, client)
+    assert isinstance(tool, StructuredTool)
+    assert tool.coroutine is not None
+    get_type_hints(tool.coroutine)