add MCP tool

Author: ks6088ts

.env.template

@@ -45,14 +45,19 @@ COSMOSDB_CONTAINER_NAME="kabuto"
 COSMOSDB_PARTITION_KEY="/id"
 
 # SQL Database Settings
-SQL_DATABASE_URI="sqlite:///template_langgraph.db"
-# SQL_DATABASE_URI="postgresql://user:password@localhost:5432/db"
+SQL_DATABASE_URI=""
+# SQL_DATABASE_URI="sqlite:///template_langgraph.db" # SQLite
+# SQL_DATABASE_URI="postgresql://user:password@localhost:5432/db" # PostgreSQL
 
 # Azure AI Search Settings
 AI_SEARCH_ENDPOINT="https://xxx.search.windows.net/"
 AI_SEARCH_KEY="xxx"
 AI_SEARCH_INDEX_NAME="kabuto"
 
+# MCP Settings
+MCP_CONFIG_PATH=""
+# MCP_CONFIG_PATH="./.vscode/mcp.json" # VS Code
+
 # ---------
 # Internals
 # ---------

.vscode/mcp.json

@@ -0,0 +1,8 @@
+{
+  "servers": {
+    "filesystem": {
+      "command": "npx",
+      "args": ["-y", "@modelcontextprotocol/server-filesystem", "./assets"]
+    }
+  }
+}

template_langgraph/tools/common.py

@@ -4,24 +4,24 @@
 from template_langgraph.tools.cosmosdb_tool import search_cosmosdb
 from template_langgraph.tools.dify_tool import run_dify_workflow
 from template_langgraph.tools.elasticsearch_tool import search_elasticsearch
+from template_langgraph.tools.mcp_tool import McpClientWrapper
 from template_langgraph.tools.qdrant_tool import search_qdrant
 from template_langgraph.tools.sql_database_tool import SqlDatabaseClientWrapper
 
 logger = get_logger(__name__)
 
 
 def get_default_tools():
-    try:
-        sql_database_tools = SqlDatabaseClientWrapper().get_tools(
+    return (
+        [
+            search_ai_search,
+            search_cosmosdb,
+            run_dify_workflow,
+            search_qdrant,
+            search_elasticsearch,
+        ]
+        + SqlDatabaseClientWrapper().get_tools(
             llm=AzureOpenAiWrapper().chat_model,
         )
-    except Exception as e:
-        logger.error(f"Error occurred while getting SQL database tools: {e}")
-        sql_database_tools = []
-    return [
-        search_ai_search,
-        search_cosmosdb,
-        run_dify_workflow,
-        search_qdrant,
-        search_elasticsearch,
-    ] + sql_database_tools
+        + McpClientWrapper().get_tools()
+    )

template_langgraph/tools/mcp_tool.py

@@ -0,0 +1,44 @@
+import asyncio
+import json
+from functools import lru_cache
+
+from langchain_core.tools.base import BaseTool
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+
+class Settings(BaseSettings):
+    mcp_config_path: str = ""
+
+    model_config = SettingsConfigDict(
+        env_file=".env",
+        env_ignore_empty=True,
+        extra="ignore",
+    )
+
+
+@lru_cache
+def get_mcp_settings() -> Settings:
+    """Get mcp settings."""
+    return Settings()
+
+
+class McpClientWrapper:
+    def __init__(
+        self,
+        settings: Settings = None,
+    ):
+        if settings is None:
+            settings = get_mcp_settings()
+        self.settings = settings
+
+    def get_tools(self) -> list[BaseTool]:
+        if self.settings.mcp_config_path == "":
+            return []
+        with open(self.settings.mcp_config_path) as f:
+            config = json.load(f)
+            for _, value in config["servers"].items():
+                value["transport"] = "stdio"
+            client = MultiServerMCPClient(config["servers"])
+            tools = asyncio.run(client.get_tools())
+            return tools

template_langgraph/tools/sql_database_tool.py

@@ -8,7 +8,7 @@
 
 
 class Settings(BaseSettings):
-    sql_database_uri: str = "sqlite:///template_langgraph.db"
+    sql_database_uri: str = ""
 
     model_config = SettingsConfigDict(
         env_file=".env",
@@ -30,15 +30,19 @@ def __init__(
     ):
         if settings is None:
             settings = get_sql_database_settings()
-        self.db = SQLDatabase.from_uri(
-            database_uri=settings.sql_database_uri,
-        )
+        self.settings = settings
 
     def get_tools(
         self,
         llm: BaseLanguageModel,
     ) -> list[BaseTool]:
         """Get SQL Database tools."""
+        if self.settings.sql_database_uri == "":
+            return []
+
+        self.db = SQLDatabase.from_uri(
+            database_uri=self.settings.sql_database_uri,
+        )
         return SQLDatabaseToolkit(
             db=self.db,
             llm=llm,