Read config from envvar

Author: jankrepl

backend/src/neuroagent/app/config.py

@@ -5,7 +5,7 @@
 from typing import Any, Literal, Optional
 
 from dotenv import dotenv_values
-from pydantic import BaseModel, ConfigDict, SecretStr, model_validator
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -241,10 +241,20 @@ def disable_if_no_url(cls, data: Any) -> Any:
         return data
 
 
+class MCPServerConfig(BaseModel):
+    """Configuration for a single MCP server."""
+
+    command: str
+    args: list[str] | None = None
+    env: dict[str, SecretStr] | None = None
+
+
 class SettingsMCP(BaseModel):
     """Settings for the MCP."""
 
-    config_path: Path | None = None
+    servers: dict[str, MCPServerConfig] = Field(
+        default_factory=dict,
+    )
 
 
 class Settings(BaseSettings):

backend/src/neuroagent/app/main.py

@@ -115,10 +115,8 @@ async def lifespan(fastapi_app: FastAPI) -> AsyncContextManager[None]:  # type:
     semantic_router = get_semantic_router(settings=app_settings)
     app.state.semantic_router = semantic_router
 
-    if app_settings.mcp.config_path is not None:
-        mcp_client: MCPClient | None = MCPClient(
-            config_path=app_settings.mcp.config_path,
-        )
+    if app_settings.mcp.servers:
+        mcp_client: MCPClient | None = MCPClient(config=app_settings.mcp)
         # start mcp servers and autogenerate input schemas
 
         await mcp_client.start()

backend/src/neuroagent/mcp.py

@@ -18,33 +18,15 @@
 from mcp.types import CallToolResult, Tool
 from neuroagent.tools.base_tool import BaseTool, BaseMetadata
 from datamodel_code_generator.parser import LiteralType
+from neuroagent.app.config import SettingsMCP
 
 logger = logging.getLogger(__name__)
 
 
-class MCPServerConfig(BaseModel):
-    command: str
-    args: list[str] | None = None
-    env: dict[str, str] | None = None
-
-
-class MCPConfig(BaseModel):
-    servers: dict[str, MCPServerConfig]
-
-    @classmethod
-    def parse_file(cls, config_path: Path) -> "MCPConfig":
-        """Parse the configuration file and return an MCPConfig object."""
-        if not config_path.exists():
-            raise FileNotFoundError(f"Configuration file {config_path} does not exist.")
-        dct = json.loads(config_path.read_text())
-
-        return cls(**dct)
-
-
 class MCPClient:
-    def __init__(self, config_path: Path):
+    def __init__(self, config: SettingsMCP):
         # Initialize session and client objects
-        self.config = MCPConfig.parse_file(config_path)
+        self.config = config
         self.exit_stack: dict[str, AsyncExitStack] = {
             name: AsyncExitStack() for name in self.config.servers.keys()
         }
@@ -79,10 +61,14 @@ async def connect_to_servers(self) -> None:
         """
         for name, server_config in self.config.servers.items():
             logger.info(f"Connecting to server: {name}")
+            if server_config.env:
+                env = {k: v.get_secret_value() for k, v in server_config.env.items()}
+            else:
+                env = None
             server_params = StdioServerParameters(
                 command=server_config.command,
                 args=server_config.args or [],
-                env=server_config.env or None,
+                env=env,
             )
 
             stdio_transport = await self.exit_stack[name].enter_async_context(