feat: Implement support for MCP Servers (Claude Tools) - Part 1 Stdio Servers

feat: (draft) support MCP Servers

feat: install npx for local MCP Servers execution

feat: add nest-asyncio as direct dependency

feat: add pdf2image to requirements.txt

feat: add local nginx for playwright file access

feat: MCP Server Support (Part 1: local stdio servers)

agent.py

@@ -1,4 +1,7 @@
 import asyncio
+import nest_asyncio
+nest_asyncio.apply()
+
 from collections import OrderedDict
 from dataclasses import dataclass, field
 from datetime import datetime
@@ -183,6 +186,25 @@ class AgentConfig:
     prompts_subdir: str = ""
     memory_subdir: str = ""
     knowledge_subdirs: list[str] = field(default_factory=lambda: ["default", "custom"])
+    mcp_servers: str = """[
+    {
+        "name": "MCP Server 1",
+        "url": "https://mcp.server.com",
+        "headers": {
+            "Authorization": "Bearer 1234567890"
+        },
+        "disabled": true,
+    },
+    {
+        "name": "MCP Server 2",
+        "command": "python3",
+        "args": ["mcp.py"],
+        "env": {
+            "PYTHONPATH": "."
+        },
+        "disabled": true,
+    }
+]"""
     code_exec_docker_enabled: bool = False
     code_exec_docker_name: str = "A0-dev"
     code_exec_docker_image: str = "frdel/agent-zero-run:development"
@@ -667,30 +689,68 @@ class Agent:
         tool_request = extract_tools.json_parse_dirty(msg)
 
         if tool_request is not None:
-            tool_name = tool_request.get("tool_name", "")
-            tool_method = None
+            raw_tool_name = tool_request.get("tool_name", "")  # Get the raw tool name
             tool_args = tool_request.get("tool_args", {})
+            
+            tool_name = raw_tool_name  # Initialize tool_name with raw_tool_name
+            tool_method = None  # Initialize tool_method
 
-            if ":" in tool_name:
-                tool_name, tool_method = tool_name.split(":", 1)
+            # Split raw_tool_name into tool_name and tool_method if applicable
+            if ":" in raw_tool_name:
+                tool_name, tool_method = raw_tool_name.split(":", 1)
+            
+            tool = None  # Initialize tool to None
 
-            tool = self.get_tool(name=tool_name, method=tool_method, args=tool_args, message=msg)
+            # Try getting tool from MCP first
+            try:
+                import python.helpers.mcp as mcp_helper 
+                mcp_tool_candidate = mcp_helper.MCPConfig.get_instance().get_tool(self, tool_name)
+                if mcp_tool_candidate:
+                    tool = mcp_tool_candidate
+            except ImportError:
+                 # Get context safely
+                 current_context = AgentContext.first()
+                 if current_context:
+                    current_context.log.log(type="warning", content="MCP helper module not found. Skipping MCP tool lookup.", temp=True)
+                 PrintStyle(background_color="black", font_color="yellow", padding=True).print(
+                    "MCP helper module not found. Skipping MCP tool lookup."
+                 )
+            except Exception as e:
+                # Get context safely
+                current_context = AgentContext.first()
+                if current_context:
+                    current_context.log.log(type="warning", content=f"Failed to get MCP tool '{tool_name}': {e}", temp=True)
+                PrintStyle(background_color="black", font_color="red", padding=True).print(
+                    f"Failed to get MCP tool '{tool_name}': {e}"
+                )
 
-            await self.handle_intervention()  # wait if paused and handle intervention message if needed
-            await tool.before_execution(**tool_args)
-            await self.handle_intervention()  # wait if paused and handle intervention message if needed
-            response = await tool.execute(**tool_args)
-            await self.handle_intervention()  # wait if paused and handle intervention message if needed
-            await tool.after_execution(response)
-            await self.handle_intervention()  # wait if paused and handle intervention message if needed
-            if response.break_loop:
-                return response.message
+            # Fallback to local get_tool if MCP tool was not found or MCP lookup failed
+            if not tool:
+                tool = self.get_tool(name=tool_name, method=tool_method, args=tool_args, message=msg)
+
+            if tool:
+                await self.handle_intervention()
+                await tool.before_execution(**tool_args)
+                await self.handle_intervention()
+                response = await tool.execute(**tool_args)
+                await self.handle_intervention()
+                await tool.after_execution(response)
+                await self.handle_intervention()
+                if response.break_loop:
+                    return response.message
+            else:
+                error_detail = f"Tool '{raw_tool_name}' not found or could not be initialized."
+                self.hist_add_warning(error_detail)
+                PrintStyle(font_color="red", padding=True).print(error_detail)
+                self.context.log.log(
+                    type="error", content=f"{self.agent_name}: {error_detail}"
+                )
         else:
-            msg = self.read_prompt("fw.msg_misformat.md")
-            self.hist_add_warning(msg)
-            PrintStyle(font_color="red", padding=True).print(msg)
+            warning_msg_misformat = self.read_prompt("fw.msg_misformat.md")
+            self.hist_add_warning(warning_msg_misformat)
+            PrintStyle(font_color="red", padding=True).print(warning_msg_misformat)
             self.context.log.log(
-                type="error", content=f"{self.agent_name}: Message misformat"
+                type="error", content=f"{self.agent_name}: Message misformat, no valid tool request found."
             )
 
     def log_from_stream(self, stream: str, logItem: Log.LogItem):

docker/run/fs/etc/nginx/nginx.conf

@@ -0,0 +1,31 @@
+daemon            off;
+worker_processes  2;
+user              www-data;
+
+events {
+    use           epoll;
+    worker_connections  128;
+}
+
+error_log         /var/log/nginx/error.log info;
+
+http {
+    server_tokens off;
+    include       mime.types;
+    charset       utf-8;
+
+    access_log    /var/log/nginx/access.log  combined;
+
+    server {
+        server_name   127.0.0.1:31735;
+        listen        127.0.0.1:31735;
+
+        error_page    500 502 503 504  /50x.html;
+
+        location      / {
+            root      /;
+        }
+
+    }
+
+}

docker/run/fs/ins/install_A0.sh

@@ -23,4 +23,4 @@ pip install torch --index-url https://download.pytorch.org/whl/cpu
 pip install -r /git/agent-zero/requirements.txt
 
 # Preload A0
-python /git/agent-zero/preload.py --dockerized=true
+python /git/agent-zero/preload.py --dockerized=true

docker/run/fs/ins/pre_install.sh

@@ -18,6 +18,7 @@ apt-get update && apt-get upgrade -y && apt-get install -y \
     wget \
     git \
     ffmpeg \
+    nginx\
     supervisor \
     cron
 
@@ -43,5 +44,8 @@ echo "=====AFTER UPDATE====="
 #   python3 -m pip install --upgrade pip
 # fi
 
+# Install npx for use by local MCP Servers
+npm i -g npx shx
+
 # Prepare SSH daemon
 bash /ins/setup_ssh.sh "$@"

initialize.py

@@ -53,6 +53,7 @@ def initialize():
         prompts_subdir=current_settings["agent_prompts_subdir"],
         memory_subdir=current_settings["agent_memory_subdir"],
         knowledge_subdirs=["default", current_settings["agent_knowledge_subdir"]],
+        mcp_servers=current_settings["mcp_servers"],
         code_exec_docker_enabled=False,
         # code_exec_docker_name = "A0-dev",
         # code_exec_docker_image = "frdel/agent-zero-run:development",
@@ -75,6 +76,23 @@ def initialize():
     # update config with runtime args
     args_override(config)
 
+    import python.helpers.mcp as mcp_helper
+    import agent as agent_helper
+    import python.helpers.print_style as print_style_helper
+    if not mcp_helper.MCPConfig.get_instance().is_initialized():
+        try:
+            mcp_helper.MCPConfig.update(config.mcp_servers)
+        except Exception as e:
+            if agent_helper.AgentContext.first():
+                (
+                    agent_helper.AgentContext.first().log
+                    .log(type="warning", content=f"Failed to update MCP settings: {e}", temp=False)
+                )
+            (
+                print_style_helper.PrintStyle(background_color="black", font_color="red", padding=True)
+                .print(f"Failed to update MCP settings: {e}")
+            )
+
     # return config object
     return config
 

prompts/default/agent.system.mcp_tools.md

@@ -0,0 +1 @@
+{{tools}}

python/api/message.py

@@ -85,4 +85,4 @@ class Message(ApiHandler):
             id=message_id,
         )
 
-        return context.communicate(UserMessage(message, attachment_paths)), context
+        return context.communicate(UserMessage(message, attachment_paths)), context

python/api/settings_set.py

@@ -3,9 +3,11 @@ from flask import Request, Response
 
 from python.helpers import settings
 
+from typing import Any
+
 
 class SetSettings(ApiHandler):
-    async def process(self, input: dict, request: Request) -> dict | Response:
+    async def process(self, input: dict[Any, Any], request: Request) -> dict[Any, Any] | Response:
         set = settings.convert_in(input)
         set = settings.set_settings(set)
         return {"settings": set}

python/extensions/system_prompt/_10_system_prompt.py

@@ -1,17 +1,22 @@
-from datetime import datetime, timezone
+from datetime import datetime
+from typing import Any, Optional
 from python.helpers.extension import Extension
+from python.helpers.mcp import MCPConfig
 from agent import Agent, LoopData
 from python.helpers.localization import Localization
 
 
 class SystemPrompt(Extension):
 
-    async def execute(self, system_prompt: list[str]=[], loop_data: LoopData = LoopData(), **kwargs):
+    async def execute(self, system_prompt: list[str] = [], loop_data: LoopData = LoopData(), **kwargs: Any):
         # append main system prompt and tools
         main = get_main_prompt(self.agent)
         tools = get_tools_prompt(self.agent)
+        mcp_tools = get_mcp_tools_prompt(self.agent)
+
         system_prompt.append(main)
         system_prompt.append(tools)
+        system_prompt.append(mcp_tools)
 
 
 def get_main_prompt(agent: Agent):
@@ -22,4 +27,8 @@ def get_tools_prompt(agent: Agent):
     prompt = agent.read_prompt("agent.system.tools.md")
     if agent.config.chat_model.vision:
         prompt += '\n' + agent.read_prompt("agent.system.tools_vision.md")
-    return prompt
+    return prompt
+
+
+def get_mcp_tools_prompt(agent: Agent):
+    return MCPConfig.get_instance().get_tools_prompt()

python/helpers/mcp.py

@@ -0,0 +1,455 @@
+from pydantic import BaseModel, Field, Discriminator, Tag, PrivateAttr
+from typing import List, Dict, Optional, Any, Union, Literal, Annotated
+from typing import (
+    List, Dict, Optional, Any,
+    Union, Literal, Annotated, ClassVar,
+)
+import threading
+import asyncio
+from contextlib import AsyncExitStack
+from shutil import which
+from mcp import ClientSession, StdioServerParameters
+from mcp.client.stdio import stdio_client
+from mcp.types import CallToolResult, ListToolsResult, JSONRPCMessage
+from anyio.streams.memory import (
+    MemoryObjectReceiveStream,
+    MemoryObjectSendStream,
+)
+from python.helpers.dirty_json import DirtyJson
+from python.helpers.print_style import PrintStyle
+import dirtyjson
+
+from python.helpers.tool import Tool, Response
+from datetime import timedelta
+
+from abc import ABC, abstractmethod
+
+
+class MCPTool(Tool):
+    """MCP Tool wrapper"""
+    async def execute(self, **kwargs: Any):
+        error = ""
+        try:
+            response: CallToolResult = await MCPConfig.get_instance().call_tool(self.name, kwargs)
+            message = "\n\n".join([item.text for item in response.content if item.type == "text"])
+            if response.isError:
+                error = message
+        except Exception as e:
+            error = f"MCP Tool Exception: {str(e)}"
+            message = f"ERROR: {str(e)}"
+
+        if error:
+            PrintStyle(
+                background_color="#CC34C3", font_color="white", bold=True, padding=True
+            ).print(f"MCPTool::Failed to call mcp tool {self.name}:")
+            PrintStyle(background_color="#AA4455", font_color="white", padding=False).print(error)
+
+            self.agent.context.log.log(
+                type="warning",
+                content=f"{self.name}: {error}",
+            )
+
+        return Response(message=message, break_loop=False)
+
+    async def before_execution(self, **kwargs: Any):
+        (
+            PrintStyle(font_color="#1B4F72", padding=True, background_color="white", bold=True)
+            .print(f"{self.agent.agent_name}: Using tool '{self.name}'")
+        )
+        self.log = self.get_log_object()
+
+        for key, value in self.args.items():
+            PrintStyle(font_color="#85C1E9", bold=True).stream(self.nice_key(key)+": ")
+            PrintStyle(font_color="#85C1E9", padding=isinstance(value, str) and "\n" in value).stream(value)
+            PrintStyle().print()
+
+    async def after_execution(self, response: Response, **kwargs: Any):
+        # Check if response or message is None
+        if not response.message.strip():
+            text = ""
+            PrintStyle(font_color="red").print(f"Warning: Tool '{self.name}' returned None response or message")
+        else:
+            text = response.message.strip()
+
+        await self.agent.hist_add_tool_result(self.name, text)
+        (
+            PrintStyle(font_color="#1B4F72", background_color="white", padding=True, bold=True)
+            .print(f"{self.agent.agent_name}: Response from tool '{self.name}'")
+        )
+        PrintStyle(font_color="#85C1E9").print(text)
+        self.log.update(content=text)
+
+
+class MCPServerRemote(BaseModel):
+    name: str = Field(default_factory=str)
+    description: Optional[str] = Field(default="Remote SSE Server")
+    url: str = Field(default_factory=str)
+    headers: dict[str, Any] | None = Field(default_factory=dict[str, Any])
+    timeout: float = 5.0
+    sse_read_timeout: float = 60.0 * 5.0
+    disabled: bool = False
+
+    __lock: ClassVar[threading.Lock] = PrivateAttr(default=threading.Lock())
+
+    def __init__(self, config: dict[str, Any]):
+        super().__init__()
+        self.update(config)
+
+    def get_tools(self) -> List[dict[str, Any]]:
+        """Get all tools from the server"""
+        return []
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Check if a tool is available"""
+        return False
+
+    async def call_tool(self, tool_name: str, input_data: Dict[str, Any]) -> CallToolResult:
+        """Call a tool with the given input data"""
+        raise NotImplementedError("MCPServerRemote does not support calling tools")
+
+    def update(self, config: dict[str, Any]) -> "MCPServerRemote":
+        with self.__lock:
+            for key, value in config.items():
+                if key in ["name", "description", "url", "headers", "timeout", "sse_read_timeout", "disabled"]:
+                    setattr(self, key, value)
+        # We already run in an event loop, dont believe Pylance
+        return asyncio.run(self.__on_update())
+
+    async def __on_update(self) -> "MCPServerRemote":
+        return self
+
+
+class MCPServerLocal(BaseModel):
+    name: str = Field(default_factory=str)
+    description: Optional[str] = Field(default="Local StdIO Server")
+    command: str = Field(default_factory=str)
+    args: list[str] = Field(default_factory=list)
+    env: dict[str, str] | None = Field(default_factory=dict[str, str])
+    encoding: str = "utf-8"
+    encoding_error_handler: Literal["strict", "ignore", "replace"] = "strict"
+    disabled: bool = False
+
+    __lock: ClassVar[threading.Lock] = PrivateAttr(default=threading.Lock())
+    __client: Optional["MCPClientLocal"] = PrivateAttr(default=None)
+
+    def __init__(self, config: dict[str, Any]):
+        super().__init__()
+        self.__client = MCPClientLocal(self)
+        self.update(config)
+
+    def get_tools(self) -> List[dict[str, Any]]:
+        """Get all tools from the server"""
+        with self.__lock:
+            return self.__client.tools
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Check if a tool is available"""
+        with self.__lock:
+            return self.__client.has_tool(tool_name)
+
+    async def call_tool(self, tool_name: str, input_data: Dict[str, Any]) -> CallToolResult:
+        """Call a tool with the given input data"""
+        with self.__lock:
+            # We already run in an event loop, dont believe Pylance
+            return await self.__client.call_tool(tool_name, input_data)
+
+    def update(self, config: dict[str, Any]) -> "MCPServerLocal":
+        with self.__lock:
+            for key, value in config.items():
+                if key in ["name", "description", "command", "args", "env", "encoding", "encoding_error_handler", "disabled"]:
+                    if key == "name":
+                        value = value.strip().lower().replace(" ", "_").replace("-", "_").replace(".", "_")
+                    setattr(self, key, value)
+            # We already run in an event loop, dont believe Pylance
+            return asyncio.run(self.__on_update())
+
+    async def __on_update(self) -> "MCPServerLocal":
+        await self.__client.update_tools()
+        return self
+
+
+MCPServer = Annotated[
+    Union[
+        Annotated[MCPServerRemote, Tag('MCPServerRemote')],
+        Annotated[MCPServerLocal, Tag('MCPServerLocal')]
+    ],
+    Discriminator(lambda v: "MCPServerRemote" if "url" in v else "MCPServerLocal")
+]
+
+
+class MCPConfig(BaseModel):
+    servers: List[MCPServer] = Field(default_factory=list[MCPServer])
+
+    __lock: ClassVar[threading.Lock] = PrivateAttr(default=threading.Lock())
+
+    # Singleton instance
+    __instance: ClassVar[Any] = PrivateAttr(default=None)
+    __initialized: ClassVar[bool] = PrivateAttr(default=False)
+
+    @classmethod
+    def get_instance(cls) -> "MCPConfig":
+        if cls.__instance is None:
+            cls.__instance = cls(servers_list=[])
+        return cls.__instance
+
+    @classmethod
+    def update(cls, config_str: str) -> Any:
+        """Parse the MCP config string into a MCPConfig object."""
+        with cls.__lock:
+            try:
+                servers = dirtyjson.loads(config_str)
+            except Exception:
+                try:
+                    servers = DirtyJson.parse_string(config_str)
+                except Exception as e:
+                    raise ValueError(f"Failed to parse MCP config: {e}")
+            cls.get_instance().__init__(servers_list=servers)
+            cls.__initialized = True
+            return cls.get_instance()
+
+    def __init__(self, servers_list: List[Dict[str, Any]]):
+        from collections.abc import Mapping, Iterable
+
+        # This empties the servers list
+        super().__init__()
+
+        if not isinstance(servers_list, Iterable):
+            (
+                PrintStyle(background_color="grey", font_color="red", padding=True)
+                .print("MCPConfig::__init__::servers_list must be a list")
+            )
+            return
+
+        for server_item in servers_list:
+            if not isinstance(server_item, Mapping):
+                (
+                    PrintStyle(background_color="grey", font_color="red", padding=True)
+                    .print("MCPConfig::__init__::server_item must be a mapping")
+                )
+                continue
+
+            if server_item.get("disabled", False):
+                continue
+
+            server_name = server_item.get("name", "__not__found__")
+            if server_name == "__not__found__":
+                (
+                    PrintStyle(background_color="grey", font_color="red", padding=True)
+                    .print("MCPConfig::__init__::server_name is required")
+                )
+                continue
+
+            try:
+                # not generic MCPServer because: "Annotated can not be instatioated"
+                if server_item.get("url", None):
+                    self.servers.append(MCPServerRemote(server_item))
+                else:
+                    self.servers.append(MCPServerLocal(server_item))
+            except Exception as e:
+                (
+                    PrintStyle(background_color="grey", font_color="red", padding=True)
+                    .print(f"MCPConfig::__init__: Failedto create MCPServer '{server_name}': {e}")
+                )
+                continue
+
+    def is_initialized(self) -> bool:
+        """Check if the client is initialized"""
+        with self.__lock:
+            return self.__initialized
+
+    def get_tools(self) -> List[dict[str, str | dict[str, Any] | None]]:
+        """Get all tools from all servers"""
+        with self.__lock:
+            tools = []
+            for server in self.servers:
+                for tool in server.get_tools():
+                    tool_copy = tool.copy()
+                    tool_copy["server"] = server.name
+                    tools.append({f"{server.name}.{tool['name']}": tool_copy})
+            return tools
+
+    def get_tools_prompt(self, server_name: str = "") -> str:
+        """Get a prompt for all tools"""
+        prompt = '## "Remote (MCP Server) Agent Tools" available:\n\n'
+        server_names = []
+        for server in self.servers:
+            if not server_name or server.name == server_name:
+                server_names.append(server.name)
+
+        if server_name and server_name not in server_names:
+            raise ValueError(f"Server {server_name} not found")
+
+        for server in self.servers:
+            if server.name in server_names:
+                server_name = server.name
+                for tool in server.get_tools():
+                    prompt += (
+                        f"### {server_name}.{tool['name']}:\n"
+                        f"{tool['description']}\n\n"
+                        f"#### Categories:\n"
+                        f"* kind: MCP Server Tool\n"
+                        f'* server: "{server_name}" ({server.description})\n\n'
+                        f"#### Arguments:\n"
+                    )
+
+                    tool_args = ""
+                    properties: dict[str, Any] = tool["input_schema"]["properties"]
+                    for key, value in properties.items():
+                        tool_args += f"            \"{key}\": \"...\",\n"
+                        if "examples" in value:
+                            prompt += (
+                                f" * {key} ({value['type']}): {value['description']} (examples: {value['examples']})\n"
+                            )
+                        else:
+                            prompt += (
+                                f" * {key} ({value['type']}): {value['description']}\n"
+                            )
+                    prompt += "\n"
+
+                    prompt += (
+                        f"#### Usage:\n"
+                        f"~~~json\n"
+                        f"{{\n"
+                        f"    \"observations\": [\"...\"],\n"
+                        f"    \"thoughts\": [\"...\"],\n"
+                        f"    \"reflection\": [\"...\"],\n"
+                        f"    \"tool_name\": \"{server_name}.{tool['name']}\",\n"
+                        f"    \"tool_args\": {{\n"
+                        f"{tool_args}"
+                        f"    }}\n"
+                        f"}}\n"
+                        f"~~~\n"
+                    )
+
+        return prompt
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Check if a tool is available"""
+        if "." not in tool_name:
+            return False
+        server_name_part, tool_name_part = tool_name.split(".")
+        with self.__lock:
+            for server in self.servers:
+                if server.name == server_name_part:
+                    return server.has_tool(tool_name_part)
+            return False
+
+    def get_tool(self, agent: Any, tool_name: str) -> MCPTool | None:
+        if not self.has_tool(tool_name):
+            return None
+        return MCPTool(agent, tool_name, {}, "", **{})
+
+    async def call_tool(self, tool_name: str, input_data: Dict[str, Any]) -> CallToolResult:
+        """Call a tool with the given input data"""
+        if "." not in tool_name:
+            raise ValueError(f"Tool {tool_name} not found")
+        server_name_part, tool_name_part = tool_name.split(".")
+        with self.__lock:
+            for server in self.servers:
+                if server.name == server_name_part and server.has_tool(tool_name_part):
+                    return await server.call_tool(tool_name_part, input_data)
+            raise ValueError(f"Tool {tool_name} not found")
+
+
+class MCPClientLocal:
+    session: Optional[ClientSession] = None
+    exit_stack: AsyncExitStack = AsyncExitStack()
+    stdio: Optional[MemoryObjectReceiveStream[JSONRPCMessage | Exception]] = None
+    write: Optional[MemoryObjectSendStream[JSONRPCMessage]] = None
+
+    tools: List[dict[str, Any]] = []
+    server: Optional[MCPServerLocal] = None
+
+    __lock: ClassVar[threading.Lock] = threading.Lock()
+
+    def __init__(self, server: MCPServerLocal):
+        self.server = server
+
+    async def __connect_to_server(self) -> Any:
+        """Connect to an MCP server"""
+
+        if not which(self.server.command):
+            raise ValueError(f"Command {self.server.command} not found")
+
+        which_args = 0
+        for arg in self.server.args:
+            if which(arg):
+                which_args = which_args + 1
+        if which_args == 0:
+            raise ValueError(f"None of the arguments {self.server.args} is a file")
+
+        with self.__lock:
+            server_params = StdioServerParameters(
+                command=self.server.command,
+                args=self.server.args,
+                env=self.server.env,
+                encoding=self.server.encoding,
+                encoding_error_handler=self.server.encoding_error_handler
+            )
+            stdio_transport = await self.exit_stack.enter_async_context(stdio_client(server_params))
+            self.stdio, self.write = stdio_transport
+
+            self.session = (
+                await self.exit_stack.enter_async_context(
+                    ClientSession(
+                        self.stdio,
+                        self.write,
+                        read_timeout_seconds=timedelta(seconds=15)
+                    )
+                )
+            )
+
+            # Initialize session
+            await self.session.initialize()
+            return self
+
+    async def update_tools(self) -> Any:
+        """List available tools from the server"""
+        try:
+            await self.__connect_to_server()
+
+            with self.__lock:
+                response: ListToolsResult = await self.session.list_tools()
+                available_tools = [{
+                        "name": tool.name,
+                        "description": tool.description,
+                        "input_schema": tool.inputSchema
+                    } for tool in response.tools]
+
+                self.tools = available_tools
+                await self.exit_stack.aclose()
+                return self
+        except Exception as e:
+            PrintStyle(
+                background_color="#CC34C3", font_color="white", bold=True, padding=True
+            ).print("MCPClientLocal::Failed to update tools:")
+            PrintStyle(background_color="#AA4455", font_color="white", padding=False).print(str(e))
+
+    def has_tool(self, tool_name: str) -> bool:
+        """Check if a tool is available"""
+        with self.__lock:
+            for tool in self.tools:
+                if tool["name"] == tool_name:
+                    return True
+        return False
+
+    def get_tools(self) -> List[dict[str, Any]]:
+        """Get all tools from the server"""
+        with self.__lock:
+            return self.tools
+
+    async def call_tool(self, tool_name: str, input_data: Dict[str, Any]) -> CallToolResult:
+        """Call a tool with the given input data"""
+        if not self.has_tool(tool_name):
+            await self.update_tools()
+
+        await self.__connect_to_server()
+
+        with self.__lock:
+            for tool in self.tools:
+                if tool["name"] == tool_name:
+                    response: CallToolResult = await self.session.call_tool(tool_name, input_data)
+                    # after connect have to close the stack within this function
+                    await self.exit_stack.aclose()
+                    return response
+            raise ValueError(f"Tool {tool_name} not found")

python/helpers/settings.py

@@ -7,6 +7,8 @@ from typing import Any, Literal, TypedDict
 import models
 from python.helpers import runtime, whisper, defer
 from . import files, dotenv
+from python.helpers.print_style import PrintStyle
+
 
 
 class Settings(TypedDict):
@@ -43,6 +45,7 @@ class Settings(TypedDict):
     agent_prompts_subdir: str
     agent_memory_subdir: str
     agent_knowledge_subdir: str
+    mcp_servers: str
 
     api_keys: dict[str, str]
 
@@ -528,6 +531,16 @@ def convert_out(settings: Settings) -> SettingsOutput:
         }
     )
 
+    agent_fields.append(
+        {
+            "id": "mcp_servers",
+            "title": "MCP Servers",
+            "description": "(JSON list of) >> RemoteServer <<: [name, url, headers, timeout (opt), sse_read_timeout (opt), disabled (opt)] / >> Local Server <<: [name, command, args, env, encoding (opt), encoding_error_handler (opt), disabled (opt)]",
+            "type": "textarea",
+            "value": settings["mcp_servers"],
+        }
+    )
+
     agent_section: SettingsSection = {
         "id": "agent",
         "title": "Agent Config",
@@ -826,6 +839,7 @@ def get_default_settings() -> Settings:
         agent_prompts_subdir="default",
         agent_memory_subdir="default",
         agent_knowledge_subdir="custom",
+        mcp_servers="",
         rfc_auto_docker=True,
         rfc_url="localhost",
         rfc_password="",
@@ -845,8 +859,9 @@ def _apply_settings(previous: Settings | None):
         from agent import AgentContext
         from initialize import initialize
 
+        config = initialize()
         for ctx in AgentContext._contexts.values():
-            ctx.config = initialize()  # reinitialize context config with new settings
+            ctx.config = config  # reinitialize context config with new settings
             # apply config to agents
             agent = ctx.agent0
             while agent:
@@ -867,6 +882,40 @@ def _apply_settings(previous: Settings | None):
             from python.helpers.memory import reload as memory_reload
             memory_reload()
 
+        # update mcp settings if necessary
+        from python.helpers.mcp import MCPConfig
+
+        async def update_mcp_settings(mcp_servers: str):
+            PrintStyle(background_color="black", font_color="white", padding=True).print("Updating MCP config...")
+            AgentContext.first().log.log(type="info", content="Updating MCP settings...", temp=True)
+
+            mcp_config = MCPConfig.get_instance()
+            try:
+                MCPConfig.update(mcp_servers)
+            except Exception as e:
+                AgentContext.first().log.log(type="warning", content=f"Failed to update MCP settings: {e}", temp=False)
+                (
+                    PrintStyle(background_color="red", font_color="black", padding=True)
+                    .print("Failed to update MCP settings")
+                )
+                (
+                    PrintStyle(background_color="black", font_color="red", padding=True)
+                    .print(f"{e}")
+                )
+
+            PrintStyle(
+                background_color="#6734C3", font_color="white", padding=True
+            ).print("Parsed MCP config:")
+            (
+                PrintStyle(background_color="#334455", font_color="white", padding=False)
+                .print(mcp_config.model_dump_json())
+            )
+            AgentContext.first().log.log(type="info", content="Finished updating MCP settings :)", temp=True)
+
+        task2 = defer.DeferredTask().start_task(
+            update_mcp_settings, config.mcp_servers
+        )  # TODO overkill, replace with background task
+
 
 def _env_to_dict(data: str):
     env_dict = {}

requirements.txt

@@ -29,4 +29,7 @@ tiktoken==0.8.0
 unstructured==0.15.13
 unstructured-client==0.25.9
 webcolors==24.6.0
+mcp==1.3.0
+nest-asyncio==1.6.0
+pdf2image==1.17.0
 crontab==1.0.1