feat(mcp): add a feature to start MCP server from config file

Author: JackYPCOnline

src/strands/tools/mcp/__init__.py

@@ -8,6 +8,7 @@
 
 from .mcp_agent_tool import MCPAgentTool
 from .mcp_client import MCPClient
+from .mcp_from_config import MCPServerConfig
 from .mcp_types import MCPTransport
 
-__all__ = ["MCPAgentTool", "MCPClient", "MCPTransport"]
+__all__ = ["MCPAgentTool", "MCPClient", "MCPServerConfig", "MCPTransport"]

src/strands/tools/mcp/mcp_from_config.py

@@ -0,0 +1,108 @@
+"""MCP server configuration loading utilities."""
+
+import json
+from pathlib import Path
+from typing import Dict, List, Optional
+
+from mcp import StdioServerParameters
+from mcp.client.stdio import stdio_client
+
+from .mcp_client import MCPClient
+from .mcp_types import MCPTransport
+
+
+class MCPServerConfig:
+    """Configuration for an MCP server following MCP standards."""
+
+    def __init__(
+        self,
+        name: str,
+        command: str,
+        args: Optional[List[str]] = None,
+        env: Optional[Dict[str, str]] = None,
+        timeout: Optional[int] = None,
+    ):
+        """Initialize MCP server configuration.
+
+        Args:
+            name: Server name
+            command: Command to run the server
+            args: Command arguments
+            env: Environment variables
+            timeout: Timeout in milliseconds
+        """
+        self.name = name
+        self.command = command
+        self.args = args or []
+        self.env = env or {}
+        self.timeout = timeout or 60000
+
+    def create_client(self) -> MCPClient:
+        """Create an MCPClient from this configuration."""
+
+        def transport_callable() -> MCPTransport:
+            server_params = StdioServerParameters(command=self.command, args=self.args, env=self.env)
+            return stdio_client(server_params)
+
+        return MCPClient(transport_callable)
+
+    @classmethod
+    def from_config(cls, config_path: str) -> List["MCPServerConfig"]:
+        """Load MCP server configurations from standard mcp.json format.
+
+        Args:
+            config_path: Path to the MCP configuration file
+
+        Returns:
+            List of MCPServerConfig instances
+
+        Config file examples:
+        Anthropic MCP Server Config Examples: (https://modelcontextprotocol.io/examples)
+        AmazonQ MCP Server Config Examples: (https://docs.aws.amazon.com/amazonq/latest/qdeveloper-ug/command-line-mcp-understanding-config.html)
+
+        Expected format:
+        {
+          "mcpServers": {
+            "server-name": {
+              "command": "command-to-run",
+              "args": ["arg1", "arg2"],
+              "env": {
+                "ENV_VAR1": "value1",
+                "ENV_VAR2": "value2"
+              },
+              "timeout": 60000
+            }
+          }
+        }
+
+        """
+        config_file = Path(config_path)
+        if not config_file.exists():
+            raise FileNotFoundError(f"Config file not found: {config_path}")
+
+        with open(config_file) as f:
+            config_data = json.load(f)
+
+        servers = []
+        mcp_server_name = set()
+        mcp_servers = config_data.get("mcpServers", {})
+        expected_attrs = {"command", "args", "env", "timeout"}
+
+        for name, server_config in mcp_servers.items():
+            if len(name) == 0 or len(name) > 250 or server_config.get("command") is None or name in mcp_server_name:
+                raise ValueError(f"Invalid server configuration for {name}")
+            if set(server_config.keys()) - expected_attrs:
+                raise ValueError(f"Invalid server configuration for {name}")
+
+            servers.append(
+                cls(
+                    name=name,
+                    command=server_config["command"],
+                    args=server_config.get("args"),
+                    env=server_config.get("env"),
+                    timeout=server_config.get("timeout"),
+                )
+            )
+            mcp_server_name.add(name)
+
+        return servers

tests/strands/tools/mcp/test_mcp_from_config.py

@@ -0,0 +1,92 @@
+import json
+from unittest.mock import patch
+
+import pytest
+
+from strands.tools.mcp.mcp_from_config import MCPServerConfig
+
+
+@pytest.fixture
+def mcp_config_data():
+    """Valid MCP configuration data."""
+    return {
+        "mcpServers": {
+            "filesystem": {
+                "command": "npx",
+                "args": ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
+                "env": {"NODE_ENV": "production"},
+                "timeout": 60000,
+            },
+            "calculator": {"command": "python", "args": ["calculator_server.py"]},
+        }
+    }
+
+
+@pytest.fixture
+def mcp_config_file(tmp_path, mcp_config_data):
+    """Create a temporary MCP config file."""
+    config_path = tmp_path / "mcp_config.json"
+    with open(config_path, "w") as f:
+        json.dump(mcp_config_data, f)
+    return str(config_path)
+
+
+@pytest.fixture
+def invalid_mcp_config_data():
+    """Invalid MCP configuration data for testing validation."""
+    return {
+        "mcpServers": {
+            "": {  # Empty name
+                "command": "echo"
+            },
+            "no_command": {  # Missing command
+                "args": ["test"]
+            },
+        }
+    }
+
+
+class TestMCPServerConfig:
+    """Test MCPServerConfig core functionality."""
+
+    def test_init_defaults(self):
+        """Test MCPServerConfig initialization with defaults."""
+        config = MCPServerConfig(name="test", command="echo")
+
+        assert config.name == "test"
+        assert config.command == "echo"
+        assert config.args == []
+        assert config.env == {}
+        assert config.timeout == 60000
+
+    @patch("strands.tools.mcp.mcp_from_config.MCPClient")
+    def test_create_client(self, mock_mcp_client):
+        """Test create_client method creates MCPClient."""
+        config = MCPServerConfig(name="test", command="echo")
+
+        result = config.create_client()
+
+        mock_mcp_client.assert_called_once()
+        assert result == mock_mcp_client.return_value
+
+    def test_from_config_valid_file(self, mcp_config_file):
+        """Test from_config with valid config file."""
+        servers = MCPServerConfig.from_config(mcp_config_file)
+
+        assert len(servers) == 2
+        server_names = {s.name for s in servers}
+        assert server_names == {"filesystem", "calculator"}
+
+    def test_from_config_file_not_found(self):
+        """Test from_config with nonexistent file."""
+        with pytest.raises(FileNotFoundError):
+            MCPServerConfig.from_config("/nonexistent/path.json")
+
+    def test_from_config_validation_error(self, tmp_path, invalid_mcp_config_data):
+        """Test from_config with validation errors."""
+        config_path = tmp_path / "invalid.json"
+        with open(config_path, "w") as f:
+            json.dump(invalid_mcp_config_data, f)
+
+        with pytest.raises(ValueError):
+            MCPServerConfig.from_config(str(config_path))