feat: Add MCP client

README.md

@@ -51,6 +51,9 @@ cat agent-spec.txt | strands "Build a specialized agent based on these specifica
 
 # Use with knowledge base to extend existing tools
 strands --kb YOUR_KB_ID "Load my previous calculator tool and enhance it with scientific functions"
+
+# Connect to MCP servers for extended capabilities
+strands --mcp-config '[{"transport": "stdio", "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem"]}]'
 ```
 
 ## Features
@@ -65,6 +68,7 @@ strands --kb YOUR_KB_ID "Load my previous calculator tool and enhance it with sc
 - 🪄 Nested agent capabilities with tool delegation
 - 🔧 Dynamic tool loading for extending functionality
 - 🖥️ Environment variable management and customization
+- 🔌 MCP (Model Context Protocol) client for connecting to external tools and services
 
 ## Integrated Tools
 
@@ -97,6 +101,7 @@ Strands comes with a comprehensive set of built-in tools:
 - **use_llm**: Run a new AI event loop with custom prompts
 - **welcome**: Manage the Strands Agent Builder welcome text
 - **workflow**: Orchestrate sequenced workflows
+- **mcp_client**: Connect to and interact with MCP (Model Context Protocol) servers
 
 ## Knowledge Base Integration
 
@@ -193,6 +198,82 @@ You can then use it with strands by running:
 $ strands --model-provider custom_model --model-config <JSON|FILE>
 ```
 
+## MCP (Model Context Protocol) Integration
+
+Strands now supports connecting to MCP servers, allowing you to extend your agent's capabilities with external tools and services. MCP provides a standardized way to connect AI assistants to various data sources and tools.
+
+### Quick Start with MCP
+
+```bash
+# Connect to a single MCP server
+strands --mcp-config '[{"transport": "stdio", "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", "/path/to/files"]}]'
+
+# Load MCP config from a file
+strands --mcp-config mcp_config.json
+
+# Use environment variable for default config
+export STRANDS_MCP_CONFIG_PATH=~/.config/mcp/servers.json
+strands
+```
+
+### MCP Configuration Format
+
+Create an `mcp_config.json` file:
+
+```json
+[
+  {
+    "connection_id": "filesystem",
+    "transport": "stdio",
+    "command": "npx",
+    "args": ["-y", "@modelcontextprotocol/server-filesystem", "/home/user/documents"],
+    "auto_load_tools": true
+  },
+  {
+    "connection_id": "github",
+    "transport": "stdio", 
+    "command": "npx",
+    "args": ["-y", "@modelcontextprotocol/server-github"],
+    "env": {
+      "GITHUB_TOKEN": "your-github-token"
+    }
+  },
+  {
+    "connection_id": "web_api",
+    "transport": "sse",
+    "server_url": "http://localhost:8080/mcp"
+  }
+]
+```
+
+### Supported MCP Transports
+
+- **stdio**: Connect to MCP servers via standard input/output
+- **sse**: Connect to MCP servers via Server-Sent Events (HTTP)
+
+### Using MCP Tools
+
+Once connected, MCP tools are automatically loaded and available to your agent:
+
+```bash
+# List available MCP connections
+strands "Show me all active MCP connections"
+
+# Use MCP filesystem tools
+strands "List all files in my documents folder"
+
+# Use MCP GitHub tools  
+strands "Create a new issue in my repository about improving documentation"
+```
+
+### MCP Connection Management
+
+Strands automatically:
+- Connects to all configured MCP servers on startup
+- Loads available tools from each server (if `auto_load_tools` is true)
+- Disconnects cleanly when exiting
+- Shows connection status during initialization
+
 ## Custom System Prompts
 
 ```bash

src/strands_agents_builder/strands.py

@@ -24,6 +24,7 @@
     image_reader,
     journal,
     load_tool,
+    mcp_client,
     memory,
     nova_reels,
     python_repl,
@@ -41,7 +42,7 @@
 from strands_tools.utils.user_input import get_user_input
 
 from strands_agents_builder.handlers.callback_handler import callback_handler
-from strands_agents_builder.utils import model_utils
+from strands_agents_builder.utils import mcp_utils, model_utils
 from strands_agents_builder.utils.kb_utils import load_system_prompt, store_conversation_in_kb
 from strands_agents_builder.utils.welcome_utils import render_goodbye_message, render_welcome_message
 
@@ -77,6 +78,12 @@ def main():
         default="{}",
         help="Model config as JSON string or path",
     )
+    parser.add_argument(
+        "--mcp-config",
+        type=mcp_utils.load_config,
+        default="[]",
+        help="MCP config as JSON string or path to JSON file. Can specify multiple MCP servers.",
+    )
     args = parser.parse_args()
 
     # Get knowledge_base_id from args or environment variable
@@ -118,6 +125,7 @@ def main():
         store_in_kb,
         strand,
         welcome,
+        mcp_client,
     ]
 
     agent = Agent(
@@ -127,6 +135,21 @@ def main():
         callback_handler=callback_handler,
     )
 
+    # Initialize MCP connections if configured
+    if args.mcp_config:
+        print("\n🔌 Initializing MCP connections...")
+        mcp_results = mcp_utils.initialize_mcp_connections(args.mcp_config, agent)
+
+        # Summary of MCP initialization
+        successful = sum(1 for success in mcp_results.values() if success)
+        total = len(mcp_results)
+        if successful == total:
+            print(f"✅ All {total} MCP connection(s) initialized successfully\n")
+        elif successful > 0:
+            print(f"⚠️  {successful}/{total} MCP connection(s) initialized\n")
+        else:
+            print("❌ Failed to initialize any MCP connections\n")
+
     # Process query or enter interactive mode
     if args.query:
         query = " ".join(args.query)
@@ -150,6 +173,10 @@ def main():
             try:
                 user_input = get_user_input("\n~ ", default="", keyboard_interrupt_return_default=False)
                 if user_input.lower() in ["exit", "quit"]:
+                    # Disconnect all MCP connections before exiting
+                    if args.mcp_config:
+                        print("\n🔌 Disconnecting MCP connections...")
+                        mcp_utils.disconnect_all(agent)
                     render_goodbye_message()
                     break
                 if user_input.startswith("!"):
@@ -189,6 +216,10 @@ def main():
                         # Store conversation in knowledge base
                         store_conversation_in_kb(agent, user_input, response, knowledge_base_id)
             except (KeyboardInterrupt, EOFError):
+                # Disconnect all MCP connections before exiting
+                if args.mcp_config:
+                    print("\n\n🔌 Disconnecting MCP connections...")
+                    mcp_utils.disconnect_all(agent)
                 render_goodbye_message()
                 break
             except Exception as e:

src/strands_agents_builder/utils/mcp_utils.py

@@ -0,0 +1,206 @@
+"""Utilities for loading and managing MCP clients in strands."""
+
+import json
+import os
+from typing import Any, Dict, List
+
+
+def load_config(config: str) -> List[Dict[str, Any]]:
+    """Load MCP configuration from a JSON string or file.
+
+    The configuration should be a list of MCP server configurations.
+    Each server configuration should contain:
+    - connection_id: Unique identifier for the connection (optional, auto-generated if missing)
+    - transport: Transport type (stdio or sse)
+    - command: Command for stdio transport (required for stdio)
+    - args: Arguments for stdio command (optional)
+    - server_url: URL for SSE transport (required for sse)
+    - auto_load_tools: Whether to automatically load tools (default: True)
+
+    Args:
+        config: A JSON string or path to a JSON file containing MCP configurations.
+            If empty string or '[]', checks STRANDS_MCP_CONFIG_PATH environment variable.
+
+    Returns:
+        List of parsed MCP server configurations.
+
+    Examples:
+        # From JSON file
+        config = load_config("mcp_config.json")
+
+        # From JSON string (connection_id is optional)
+        config = load_config('[{"transport": "stdio", "command": "node", "args": ["server.js"]}]')
+    """
+    if not config or config == "[]":
+        # Check for default config path in environment
+        default_path = os.getenv("STRANDS_MCP_CONFIG_PATH")
+        if default_path and os.path.exists(default_path):
+            config = default_path
+        else:
+            return []
+
+    if config.endswith(".json"):
+        with open(config) as fp:
+            data = json.load(fp)
+    else:
+        data = json.loads(config)
+
+    # Handle Amazon Q MCP format
+    if isinstance(data, dict) and "mcpServers" in data:
+        servers = []
+        for server_id, server_config in data["mcpServers"].items():
+            # Skip disabled servers
+            if server_config.get("disabled", False):
+                continue
+
+            # Convert to our format
+            converted = {
+                "connection_id": server_id,
+                "transport": "stdio",  # Amazon Q format uses stdio by default
+                "command": server_config.get("command"),
+                "args": server_config.get("args", []),
+                "auto_load_tools": True,
+            }
+
+            # Add environment variables if present
+            if "env" in server_config:
+                converted["env"] = server_config["env"]
+
+            servers.append(converted)
+        data = servers
+
+    # Ensure it's a list
+    if isinstance(data, dict):
+        data = [data]
+
+    return data
+
+
+def initialize_mcp_connections(configs: List[Dict[str, Any]], agent) -> Dict[str, bool]:
+    """Initialize MCP connections based on provided configurations.
+
+    Args:
+        configs: List of MCP server configurations.
+        agent: The strands agent instance to use for MCP client calls.
+
+    Returns:
+        Dictionary mapping connection_id to success status.
+    """
+    results = {}
+
+    for i, config in enumerate(configs):
+        connection_id = config.get("connection_id")
+        if not connection_id:
+            # Auto-generate connection ID based on transport and command/url
+            transport = config.get("transport", "stdio")
+            if transport == "stdio":
+                command = config.get("command", "unknown")
+                # Use the command name as basis for ID
+                base_name = os.path.basename(command).replace(".", "_")
+                connection_id = f"mcp_{base_name}_{i}"
+            else:  # sse
+                server_url = config.get("server_url", "")
+                # Extract hostname or use index
+                try:
+                    from urllib.parse import urlparse
+
+                    parsed = urlparse(server_url)
+                    host = parsed.hostname or "server"
+                    host = host.replace(".", "_").replace("-", "_")
+                    connection_id = f"mcp_{host}_{i}"
+                except Exception:
+                    connection_id = f"mcp_sse_{i}"
+
+            print(f"📝 Auto-generated connection_id: {connection_id}")
+            config["connection_id"] = connection_id
+
+        try:
+            # Connect to the MCP server
+            connect_params = {"action": "connect", "connection_id": connection_id, "kwargs": {}}
+
+            # Add transport-specific parameters
+            if "transport" in config:
+                connect_params["transport"] = config["transport"]
+
+            if config.get("transport") == "stdio":
+                if "command" in config:
+                    connect_params["command"] = config["command"]
+                if "args" in config:
+                    connect_params["args"] = config["args"]
+                if "env" in config:
+                    connect_params["env"] = config["env"]
+            elif config.get("transport") == "sse":
+                if "server_url" in config:
+                    connect_params["server_url"] = config["server_url"]
+
+            # Connect to the server
+            result = agent.tool.mcp_client(**connect_params)
+
+            if result.get("status") == "success":
+                print(f"✓ Connected to MCP server: {connection_id}")
+
+                # Auto-load tools if specified (default: True)
+                if config.get("auto_load_tools", True):
+                    load_result = agent.tool.mcp_client(action="load_tools", connection_id=connection_id, kwargs={})
+                    if load_result.get("status") == "success":
+                        print(f"  ✓ Loaded tools from {connection_id}")
+                    else:
+                        print(f"  ✗ Failed to load tools from {connection_id}")
+
+                results[connection_id] = True
+            else:
+                print(f"✗ Failed to connect to MCP server: {connection_id}")
+                error_msg = result.get("content", [{}])[0].get("text", "Unknown error")
+                print(f"  Error: {error_msg}")
+                results[connection_id] = False
+
+        except Exception as e:
+            print(f"✗ Error connecting to MCP server {connection_id}: {str(e)}")
+            results[connection_id] = False
+
+    return results
+
+
+def list_active_connections(agent) -> List[str]:
+    """List all active MCP connections.
+
+    Args:
+        agent: The strands agent instance to use for MCP client calls.
+
+    Returns:
+        List of active connection IDs.
+    """
+    try:
+        result = agent.tool.mcp_client(action="list_connections", kwargs={})
+        if result.get("status") == "success":
+            # Parse the response to extract connection IDs
+            content = result.get("content", [{}])[0].get("text", "")
+            if "No active MCP connections" in content:
+                return []
+
+            # Extract connection IDs from the formatted output
+            connections = []
+            lines = content.split("\n")
+            for line in lines:
+                if "Connection:" in line:
+                    conn_id = line.split("Connection:")[1].strip()
+                    connections.append(conn_id)
+            return connections
+        return []
+    except Exception:
+        return []
+
+
+def disconnect_all(agent) -> None:
+    """Disconnect all active MCP connections.
+
+    Args:
+        agent: The strands agent instance to use for MCP client calls.
+    """
+    connections = list_active_connections(agent)
+    for connection_id in connections:
+        try:
+            agent.tool.mcp_client(action="disconnect", connection_id=connection_id, kwargs={})
+            print(f"✓ Disconnected from MCP server: {connection_id}")
+        except Exception as e:
+            print(f"✗ Error disconnecting from {connection_id}: {str(e)}")