Strands MCP Server

Instructions

Implementation Reference

• strands_mcp_server/mcp_client.py:52-140 (handler)

  The primary handler function for the 'mcp_client' tool, decorated with @tool. It dispatches based on the 'action' parameter to connect/disconnect/list/call tools on remote MCP servers.
  @tool def mcp_client( action: str, connection_id: Optional[str] = None, transport: Optional[str] = None, server_url: Optional[str] = None, command: Optional[str] = None, args: Optional[list[str]] = None, tool_name: Optional[str] = None, tool_args: Optional[dict[str, Any]] = None, ) -> dict[str, Any]: """Test and interact with MCP servers. This tool provides a complete MCP client implementation for testing and using MCP servers from within a Strands Agent. Args: action: Action to perform - "connect", "disconnect", "list_tools", "call_tool", "list_connections" connection_id: Unique identifier for this connection transport: Transport type - "http", "stdio", or "sse" server_url: URL for HTTP/SSE transport (e.g., "http://localhost:8000/mcp") command: Command for stdio transport (e.g., "python") args: Arguments for stdio command (e.g., ["mcp_server_stdio.py"]) tool_name: Name of tool to call (for call_tool action) tool_args: Arguments to pass to tool (for call_tool action) Returns: Result dictionary with status and content Examples: # Connect to HTTP server mcp_client( action="connect", connection_id="my-server", transport="http", server_url="http://localhost:8000/mcp" ) # Connect to stdio server mcp_client( action="connect", connection_id="stdio-server", transport="stdio", command="python", args=["mcp_server_stdio.py"] ) # List tools from connection mcp_client(action="list_tools", connection_id="my-server") # Call a tool mcp_client( action="call_tool", connection_id="my-server", tool_name="calculator", tool_args={"expression": "2 + 2"} ) # List all connections mcp_client(action="list_connections") # Disconnect mcp_client(action="disconnect", connection_id="my-server") Notes: - stdio transport: Server must be launchable as subprocess - HTTP transport: Server must be already running - Connections are maintained in global state for reuse """ if action == "connect": return _connect(connection_id, transport, server_url, command, args) elif action == "disconnect": return _disconnect(connection_id) elif action == "list_tools": return _list_tools(connection_id) elif action == "call_tool": return _call_tool(connection_id, tool_name, tool_args) elif action == "list_connections": return _list_connections() else: return { "status": "error", "content": [ { "text": f"❌ Unknown action: {action}\n\n" "Available actions: connect, disconnect, list_tools, call_tool, list_connections" } ], }
• strands_mcp_server/mcp_client.py:143-309 (helper)

  Core helper implementing connection establishment for different transports, session testing, and global state management.
  def _connect( connection_id: Optional[str], transport: Optional[str], server_url: Optional[str], command: Optional[str], args: Optional[list[str]], ) -> dict[str, Any]: """Connect to an MCP server. This function establishes a connection to an MCP server using the specified transport type. The connection is stored in global state for reuse. Follows patterns from: - Strands MCPClient: sdk-python/src/strands/tools/mcp/mcp_client.py - MCP client transports: python-sdk/src/mcp/client/ """ try: if not connection_id: return { "status": "error", "content": [{"text": "❌ connection_id is required"}], } if not transport: return { "status": "error", "content": [{"text": "❌ transport is required (http, stdio, or sse)"}], } if connection_id in _client_connections: return { "status": "error", "content": [{"text": f"❌ Connection '{connection_id}' already exists"}], } # Import MCP client components from mcp.client.session import ClientSession from mcp.client.stdio import StdioServerParameters, stdio_client from mcp.client.streamable_http import streamablehttp_client if transport == "http": if not server_url: return { "status": "error", "content": [{"text": "❌ server_url is required for HTTP transport"}], } logger.debug(f"Connecting to HTTP server: {server_url}") # Create transport callable following Strands MCPClient pattern def transport_callable(): return streamablehttp_client(server_url) connection_info = { "transport": "http", "server_url": server_url, "transport_callable": transport_callable, "session": None, # Will be initialized on first use } elif transport == "stdio": if not command: return { "status": "error", "content": [{"text": "❌ command is required for stdio transport"}], } logger.debug(f"Connecting to stdio server: {command} {args or []}") # Create stdio server parameters server_params = StdioServerParameters(command=command, args=args or [], env=None) # Create transport callable def transport_callable(): return stdio_client(server_params) connection_info = { "transport": "stdio", "command": command, "args": args, "transport_callable": transport_callable, "session": None, } elif transport == "sse": if not server_url: return { "status": "error", "content": [{"text": "❌ server_url is required for SSE transport"}], } logger.debug(f"Connecting to SSE server: {server_url}") from mcp.client.sse import sse_client def transport_callable(): return sse_client(server_url) connection_info = { "transport": "sse", "server_url": server_url, "transport_callable": transport_callable, "session": None, } else: return { "status": "error", "content": [{"text": f"❌ Unknown transport: {transport}\n\nSupported: http, stdio, sse"}], } # Test the connection by listing tools logger.debug(f"Testing connection by listing tools...") tools = [] try: # Create a temporary session to test and get tool count import asyncio async def test_connection(): async with connection_info["transport_callable"]() as ( read_stream, write_stream, get_session_id, # StreamableHTTP returns 3 values ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() result = await session.list_tools() return result.tools tools = asyncio.run(test_connection()) logger.debug(f"Successfully connected, found {len(tools)} tools") except Exception as e: logger.exception("Failed to connect to MCP server") return { "status": "error", "content": [{"text": f"❌ Failed to connect: {str(e)}"}], } # Store connection _client_connections[connection_id] = connection_info # Build response tool_list = "\n".join(f" • {tool.name}" for tool in tools[:10]) if len(tools) > 10: tool_list += f"\n ... and {len(tools) - 10} more" transport_info = ( f"URL: {server_url}" if transport in ["http", "sse"] else f"Command: {command} {' '.join(args or [])}" ) message = ( f"✅ Connected to MCP server '{connection_id}'\n\n" f"📊 Transport: {transport}\n" f"🔗 {transport_info}\n" f"🔧 Available tools ({len(tools)}):\n" f"{tool_list}" ) return {"status": "success", "content": [{"text": message}]} except Exception as e: logger.exception("Error in connect action") return { "status": "error", "content": [{"text": f"❌ Error: {str(e)}"}], }
• strands_mcp_server/mcp_client.py:348-414 (helper)

  Helper to list available tools from a connected MCP server, including descriptions and input schemas.
  def _list_tools(connection_id: Optional[str]) -> dict[str, Any]: """List tools from a connected MCP server. This creates a temporary session to query the server for its available tools. """ try: if not connection_id: return { "status": "error", "content": [{"text": "❌ connection_id is required"}], } if connection_id not in _client_connections: return { "status": "error", "content": [{"text": f"❌ Connection '{connection_id}' not found"}], } connection_info = _client_connections[connection_id] logger.debug(f"Listing tools from '{connection_id}'") # Create session and list tools import asyncio from mcp.client.session import ClientSession async def list_tools_async(): async with connection_info["transport_callable"]() as ( read_stream, write_stream, get_session_id, # StreamableHTTP returns 3 values ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() result = await session.list_tools() return result.tools tools = asyncio.run(list_tools_async()) logger.debug(f"Found {len(tools)} tools") # Build detailed tool list tool_details = [] for tool in tools: details = f"**{tool.name}**" if tool.description: details += f"\n Description: {tool.description}" if tool.inputSchema: # Show required parameters schema = tool.inputSchema if "required" in schema: details += f"\n Required: {', '.join(schema['required'])}" if "properties" in schema: details += f"\n Parameters: {', '.join(schema['properties'].keys())}" tool_details.append(details) tools_text = "\n\n".join(tool_details) message = f"📋 **Tools from '{connection_id}'**\n\n" f"Found {len(tools)} tools:\n\n" f"{tools_text}" return {"status": "success", "content": [{"text": message}]} except Exception as e: logger.exception("Error listing tools") return { "status": "error", "content": [{"text": f"❌ Error: {str(e)}"}], }
• strands_mcp_server/mcp_client.py:417-487 (helper)

  Helper to execute remote tools on connected MCP servers via ClientSession.call_tool().
  def _call_tool( connection_id: Optional[str], tool_name: Optional[str], tool_args: Optional[dict[str, Any]], ) -> dict[str, Any]: """Call a tool on a connected MCP server. This establishes a session, calls the specified tool with provided arguments, and returns the result. """ try: if not connection_id: return { "status": "error", "content": [{"text": "❌ connection_id is required"}], } if not tool_name: return { "status": "error", "content": [{"text": "❌ tool_name is required"}], } if connection_id not in _client_connections: return { "status": "error", "content": [{"text": f"❌ Connection '{connection_id}' not found"}], } connection_info = _client_connections[connection_id] logger.debug(f"Calling tool '{tool_name}' on '{connection_id}' with args: {tool_args}") # Call the tool import asyncio from mcp.client.session import ClientSession async def call_tool_async(): async with connection_info["transport_callable"]() as ( read_stream, write_stream, get_session_id, ): async with ClientSession(read_stream, write_stream) as session: await session.initialize() result = await session.call_tool(tool_name, tool_args or {}) return result result = asyncio.run(call_tool_async()) logger.debug(f"Tool call complete, got {len(result.content)} content items") # Extract result content result_text = [] for content in result.content: if hasattr(content, "text"): result_text.append(content.text) else: result_text.append(str(content)) combined_result = "\n".join(result_text) message = f"✅ **Tool '{tool_name}' executed on '{connection_id}'**\n\n" f"Result:\n{combined_result}" return {"status": "success", "content": [{"text": message}]} except Exception as e: logger.exception(f"Error calling tool '{tool_name}'") return { "status": "error", "content": [{"text": f"❌ Error: {str(e)}"}], }
• strands_mcp_server/__init__.py:199-203 (registration)

  Package-level registration: imports mcp_client and adds to __all__, making it directly importable as 'from strands_mcp_server import mcp_client' for use in Agents.
  from strands_mcp_server.mcp_client import mcp_client from strands_mcp_server.mcp_server import mcp_server __version__ = "0.1.3" __all__ = ["mcp_server", "mcp_client"]