fix: don't throw an error if adding a new mcp server, also allow sett… (#1293)

letta/server/server.py

@@ -1266,6 +1266,11 @@ class SyncServer(Server):
     # TODO support both command + SSE servers (via config)
     def get_mcp_servers(self) -> dict[str, Union[SSEServerConfig, StdioServerConfig]]:
         """List the MCP servers in the config (doesn't test that they are actually working)"""
+
+        # TODO implement non-flatfile mechanism
+        if not tool_settings.mcp_read_from_config:
+            raise RuntimeError("MCP config file disabled. Enable it in settings.")
+
         mcp_server_list = {}
 
         # Attempt to read from ~/.letta/mcp_config.json
@@ -1329,10 +1334,15 @@ class SyncServer(Server):
     ) -> dict[str, Union[SSEServerConfig, StdioServerConfig]]:
         """Add a new server config to the MCP config file"""
 
+        # TODO implement non-flatfile mechanism
+        if not tool_settings.mcp_read_from_config:
+            raise RuntimeError("MCP config file disabled. Enable it in settings.")
+
         # If the config file doesn't exist, throw an error.
         mcp_config_path = os.path.join(constants.LETTA_DIR, constants.MCP_CONFIG_NAME)
         if not os.path.exists(mcp_config_path):
-            raise FileNotFoundError(f"MCP config file not found: {mcp_config_path}")
+            # Create the file if it doesn't exist
+            logger.debug(f"MCP config file not found, creating new file at: {mcp_config_path}")
 
         # If the file does exist, attempt to parse it get calling get_mcp_servers
         try:
@@ -1384,9 +1394,14 @@ class SyncServer(Server):
     def delete_mcp_server_from_config(self, server_name: str) -> dict[str, Union[SSEServerConfig, StdioServerConfig]]:
         """Delete a server config from the MCP config file"""
 
+        # TODO implement non-flatfile mechanism
+        if not tool_settings.mcp_read_from_config:
+            raise RuntimeError("MCP config file disabled. Enable it in settings.")
+
         # If the config file doesn't exist, throw an error.
         mcp_config_path = os.path.join(constants.LETTA_DIR, constants.MCP_CONFIG_NAME)
         if not os.path.exists(mcp_config_path):
+            # If the file doesn't exist, raise an error
             raise FileNotFoundError(f"MCP config file not found: {mcp_config_path}")
 
         # If the file does exist, attempt to parse it get calling get_mcp_servers

letta/settings.py

@@ -18,6 +18,9 @@ class ToolSettings(BaseSettings):
     # Local Sandbox configurations
     local_sandbox_dir: Optional[str] = None
 
+    # MCP read from config file
+    mcp_read_from_config: bool = True  # if False, will throw if attempting to read/write from file
+
 
 class SummarizerSettings(BaseSettings):
     model_config = SettingsConfigDict(env_prefix="letta_summarizer_", extra="ignore")

tests/test_client.py

@@ -1,14 +1,12 @@
 import asyncio
-import json
 import os
 import threading
 import time
 import uuid
-from typing import List, Union
 
 import pytest
 from dotenv import load_dotenv
-from letta_client import AgentState, JobStatus, Letta, MessageCreate, MessageRole
+from letta_client import AgentState, Letta, MessageCreate
 from letta_client.core.api_error import ApiError
 from sqlalchemy import delete
 

tests/test_streaming.py

@@ -5,8 +5,6 @@ import time
 import pytest
 from dotenv import load_dotenv
 from letta_client import AgentState, Letta, LlmConfig, MessageCreate
-from letta_client.core.api_error import ApiError
-from pytest import fixture
 
 
 def run_server():