Strands MCP Server

Instructions

# 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")

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"]