"""Load MCP tools using langchain-mcp-adapters.""" import logging from langchain_core.tools import BaseTool from src.config.extensions_config import ExtensionsConfig from src.mcp.client import build_servers_config logger = logging.getLogger(__name__) async def get_mcp_tools() -> list[BaseTool]: """Get all tools from enabled MCP servers. Returns: List of LangChain tools from all enabled MCP servers. """ try: from langchain_mcp_adapters.client import MultiServerMCPClient except ImportError: logger.warning("langchain-mcp-adapters not installed. Install it to enable MCP tools: pip install langchain-mcp-adapters") return [] # NOTE: We use ExtensionsConfig.from_file() instead of get_extensions_config() # to always read the latest configuration from disk. This ensures that changes # made through the Gateway API (which runs in a separate process) are immediately # reflected when initializing MCP tools. extensions_config = ExtensionsConfig.from_file() servers_config = build_servers_config(extensions_config) if not servers_config: logger.info("No enabled MCP servers configured") return [] try: # Create the multi-server MCP client logger.info(f"Initializing MCP client with {len(servers_config)} server(s)") client = MultiServerMCPClient(servers_config) # Get all tools from all servers tools = await client.get_tools() logger.info(f"Successfully loaded {len(tools)} tool(s) from MCP servers") return tools except Exception as e: logger.error(f"Failed to load MCP tools: {e}", exc_info=True) return []