feat: add get_info method to retrieve MCP server metadata

langchain_mcp_adapters/client.py

@@ -13,12 +13,11 @@
 from langchain_core.documents.base import Blob
 from langchain_core.messages import AIMessage, HumanMessage
 from langchain_core.tools import BaseTool
-from mcp import ClientSession
-
 from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks
 from langchain_mcp_adapters.interceptors import ToolCallInterceptor
 from langchain_mcp_adapters.prompts import load_mcp_prompt
 from langchain_mcp_adapters.resources import load_mcp_resources
+from langchain_mcp_adapters.server_info import load_mcp_server_info
 from langchain_mcp_adapters.sessions import (
     Connection,
     McpHttpClientFactory,
@@ -29,6 +28,8 @@
     create_session,
 )
 from langchain_mcp_adapters.tools import load_mcp_tools
+from mcp import ClientSession
+from mcp.types import InitializeResult
 
 ASYNC_CONTEXT_MANAGER_ERROR = (
     "As of langchain-mcp-adapters 0.1.0, MultiServerMCPClient cannot be used as a "
@@ -199,6 +200,29 @@ async def get_tools(self, *, server_name: str | None = None) -> list[BaseTool]:
             all_tools.extend(tools)
         return all_tools
 
+    async def get_info(self) -> dict[str, InitializeResult]:
+        """Get server info from all connected MCP servers.
+
+        Returns the ``InitializeResult`` for each server, which includes
+        server instructions, capabilities, and implementation details.
+
+        Returns:
+            A dict mapping server names to their ``InitializeResult``.
+
+        """
+        tasks = [
+            asyncio.create_task(
+                load_mcp_server_info(
+                    connection=self.connections[name],
+                    callbacks=self.callbacks,
+                    server_name=name,
+                )
+            )
+            for name in self.connections
+        ]
+        results = await asyncio.gather(*tasks)
+        return dict(zip(self.connections.keys(), results))
+
     async def get_prompt(
         self,
         server_name: str,

langchain_mcp_adapters/server_info.py

@@ -0,0 +1,65 @@
+"""Server info adapter for retrieving MCP server metadata.
+
+This module provides functionality to retrieve server information
+from the MCP initialize handshake, including server instructions,
+capabilities, and implementation details.
+"""
+
+from mcp import ClientSession
+from mcp.types import InitializeResult
+
+from langchain_mcp_adapters.callbacks import CallbackContext, Callbacks, _MCPCallbacks
+from langchain_mcp_adapters.sessions import Connection, create_session
+
+
+async def load_mcp_server_info(
+    session: ClientSession | None = None,
+    *,
+    connection: Connection | None = None,
+    callbacks: Callbacks | None = None,
+    server_name: str | None = None,
+) -> InitializeResult:
+    """Load server info from the MCP initialize handshake.
+
+    Returns the full `InitializeResult` from the MCP protocol, which includes
+    server instructions, capabilities, implementation details, and protocol
+    version.
+
+    Args:
+        session: An **uninitialized** MCP client session. If provided, this
+            function will call ``initialize()`` on it. If ``None``, a
+            ``connection`` must be provided and a temporary session will be
+            created automatically.
+        connection: Connection config to create a new session if ``session`` is
+            ``None``.
+        callbacks: Optional ``Callbacks`` for handling notifications and events.
+        server_name: Name of the server (used for callback context).
+
+    Returns:
+        The ``InitializeResult`` from the MCP server, containing:
+            - ``instructions``: Optional server instructions for the LLM.
+            - ``serverInfo``: Server implementation details (name, version).
+            - ``capabilities``: Server capabilities.
+            - ``protocolVersion``: MCP protocol version.
+
+    Raises:
+        ValueError: If neither ``session`` nor ``connection`` is provided.
+
+    """
+    if session is not None:
+        return await session.initialize()
+
+    if connection is None:
+        msg = "Either a session or a connection config must be provided"
+        raise ValueError(msg)
+
+    mcp_callbacks = (
+        callbacks.to_mcp_format(context=CallbackContext(server_name=server_name))
+        if callbacks is not None
+        else _MCPCallbacks()
+    )
+
+    async with create_session(
+        connection, mcp_callbacks=mcp_callbacks
+    ) as new_session:
+        return await new_session.initialize()

tests/test_server_info.py

@@ -0,0 +1,113 @@
+from unittest.mock import AsyncMock
+
+import pytest
+from mcp.server import FastMCP
+from mcp.types import InitializeResult
+
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from langchain_mcp_adapters.server_info import load_mcp_server_info
+from tests.utils import run_streamable_http
+
+
+def _create_server_with_instructions():
+    server = FastMCP(
+        "test-server",
+        instructions="Use this server for testing purposes only.",
+        port=8187,
+    )
+
+    @server.tool()
+    def ping() -> str:
+        """Ping the server."""
+        return "pong"
+
+    return server
+
+
+def _create_server_without_instructions():
+    server = FastMCP("no-instructions-server", port=8188)
+
+    @server.tool()
+    def ping() -> str:
+        """Ping the server."""
+        return "pong"
+
+    return server
+
+
+async def test_load_mcp_server_info_with_connection(socket_enabled) -> None:
+    """Test loading server info using a connection config."""
+    with run_streamable_http(_create_server_with_instructions, 8187):
+        result = await load_mcp_server_info(
+            connection={
+                "url": "http://localhost:8187/mcp",
+                "transport": "streamable_http",
+            },
+        )
+        assert isinstance(result, InitializeResult)
+        assert result.instructions == "Use this server for testing purposes only."
+        assert result.serverInfo.name == "test-server"
+
+
+async def test_load_mcp_server_info_no_instructions(socket_enabled) -> None:
+    """Test loading server info when server has no instructions."""
+    with run_streamable_http(_create_server_without_instructions, 8188):
+        result = await load_mcp_server_info(
+            connection={
+                "url": "http://localhost:8188/mcp",
+                "transport": "streamable_http",
+            },
+        )
+        assert isinstance(result, InitializeResult)
+        assert result.instructions is None
+
+
+async def test_load_mcp_server_info_with_session() -> None:
+    """Test loading server info with an uninitialized session."""
+    mock_result = InitializeResult(
+        protocolVersion="2025-03-26",
+        capabilities={},
+        serverInfo={"name": "mock-server", "version": "1.0"},
+        instructions="Mock instructions",
+    )
+    session = AsyncMock()
+    session.initialize.return_value = mock_result
+
+    result = await load_mcp_server_info(session)
+
+    session.initialize.assert_called_once()
+    assert result.instructions == "Mock instructions"
+    assert result.serverInfo.name == "mock-server"
+
+
+async def test_load_mcp_server_info_raises_without_args() -> None:
+    """Test that ValueError is raised when neither session nor connection."""
+    with pytest.raises(ValueError, match="Either a session or a connection"):
+        await load_mcp_server_info()
+
+
+async def test_client_get_server_info(socket_enabled) -> None:
+    """Test MultiServerMCPClient.get_server_info returns info for all servers."""
+    with (
+        run_streamable_http(_create_server_with_instructions, 8187),
+        run_streamable_http(_create_server_without_instructions, 8188),
+    ):
+        client = MultiServerMCPClient(
+            {
+                "with_instructions": {
+                    "url": "http://localhost:8187/mcp",
+                    "transport": "streamable_http",
+                },
+                "without_instructions": {
+                    "url": "http://localhost:8188/mcp",
+                    "transport": "streamable_http",
+                },
+            },
+        )
+        info = await client.get_info()
+        assert len(info) == 2
+        assert info["with_instructions"].instructions == (
+            "Use this server for testing purposes only."
+        )
+        assert info["with_instructions"].serverInfo.name == "test-server"
+        assert info["without_instructions"].instructions is None