File size: 2,652 Bytes
033ca06 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 | """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
from src.mcp.oauth import build_oauth_tool_interceptor, get_initial_oauth_headers
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)")
# Inject initial OAuth headers for server connections (tool discovery/session init)
initial_oauth_headers = await get_initial_oauth_headers(extensions_config)
for server_name, auth_header in initial_oauth_headers.items():
if server_name not in servers_config:
continue
if servers_config[server_name].get("transport") in ("sse", "http"):
existing_headers = dict(servers_config[server_name].get("headers", {}))
existing_headers["Authorization"] = auth_header
servers_config[server_name]["headers"] = existing_headers
tool_interceptors = []
oauth_interceptor = build_oauth_tool_interceptor(extensions_config)
if oauth_interceptor is not None:
tool_interceptors.append(oauth_interceptor)
client = MultiServerMCPClient(servers_config, tool_interceptors=tool_interceptors)
# 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 []
|