Sample MCP client to execute tools

test/.env

@@ -0,0 +1,10 @@
+# MCP Server details
+MCP_SERVER=127.0.0.1
+MCP_PORT=30030
+
+# Device Details (for gather_device_facts)
+DEVICE_NAME_FOR_GATHER_DEVICE_FACTS=
+
+# Device Details (for set user password)
+DEVICE_NAME_FOR_SET_USER_PASSWORD=
+PLAIN_TEXT_PASSWORD_FOR_SET_USER_PASSWORD=

test/test_mcp_client_gather_device_facts.py

@@ -0,0 +1,47 @@
+import asyncio
+import os
+import sys
+from dotenv import load_dotenv
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+# Get value for the specified variable from .env
+def get_env_variable(var_name: str) -> str:
+    """Fetch env var or exit with error if missing."""
+    value = os.getenv(var_name)
+    if value is None or value.strip() == "":
+        sys.exit(f"Missing required environment variable: {var_name}")
+    return value
+
+async def main():
+    # Load environment variables from .env
+    load_dotenv()
+
+    MCP_SERVER = get_env_variable("MCP_SERVER")
+    MCP_PORT = get_env_variable("MCP_PORT")
+
+    MCP_URL = f"http://{MCP_SERVER}:{MCP_PORT}/mcp"
+
+    DEVICE_NAME = get_env_variable("DEVICE_NAME_FOR_GATHER_DEVICE_FACTS")
+
+    # Connect to the streamable HTTP MCP server
+    async with streamablehttp_client(MCP_URL) as (read_stream, write_stream, _):
+        # Wrap the streams in a ClientSession
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the session
+            await session.initialize()
+
+            # Call "gather_device_facts" on the specified device
+            tool_name = "gather_device_facts"
+            tool_args = {"router_name": DEVICE_NAME}
+
+            print(f"\nCalling tool: {tool_name} with arguments: {tool_args}")
+            try:
+                result = await session.call_tool(name=tool_name, arguments=tool_args)
+                print("Tool execution result:")
+                print(result)
+            except Exception as e:
+                print(f"Error calling MCP tool: {e}")
+
+if __name__ == "__main__":
+    asyncio.run(main())

test/test_mcp_client_list_tools.py

@@ -0,0 +1,43 @@
+import asyncio
+import os
+from dotenv import load_dotenv
+
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+# Get value for the specified variable from .env
+def get_env_variable(var_name: str) -> str:
+    """Fetch env var or exit with error if missing."""
+    value = os.getenv(var_name)
+    if value is None or value.strip() == "":
+        sys.exit(f"Missing required environment variable: {var_name}")
+    return value
+
+async def main():
+    # Load environment variables from .env
+    load_dotenv()
+
+    MCP_SERVER = get_env_variable("MCP_SERVER")
+    MCP_PORT = get_env_variable("MCP_PORT")
+
+    MCP_URL = f"http://{MCP_SERVER}:{MCP_PORT}/mcp"
+
+    # Connect to a streamable HTTP server
+    async with streamablehttp_client(MCP_URL) as (
+        read_stream,
+        write_stream,
+        _,
+    ):
+        # Create a session using the client streams
+        async with ClientSession(read_stream, write_stream) as session:
+            # Initialize the connection
+            await session.initialize()
+            # List available tools
+            tools = await session.list_tools()
+            print("Available tools:")
+            for tool in tools.tools:
+              print(f"- {tool.name}: {tool.description}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

test/test_mcp_client_set_user_password.py

@@ -0,0 +1,70 @@
+import crypt
+import asyncio
+import os
+from dotenv import load_dotenv
+from mcp import ClientSession
+from mcp.client.streamable_http import streamablehttp_client
+
+# Get value for the specified variable from .env
+def get_env_variable(var_name: str) -> str:
+    """Fetch env var or exit with error if missing."""
+    value = os.getenv(var_name)
+    if value is None or value.strip() == "":
+        sys.exit(f"Missing required environment variable: {var_name}")
+    return value
+
+# Generate SHA-512 password hash for Junos
+def generate_junos_hash(password: str) -> str:
+    # $6$ means SHA-512, crypt.gensalt() generates a random salt
+    salt = crypt.mksalt(crypt.METHOD_SHA512)
+    return crypt.crypt(password, salt)
+
+# Execute the tool on the junos MCP server
+async def main():
+    # Load environment variables from .env
+    load_dotenv()
+
+    MCP_SERVER = get_env_variable("MCP_SERVER")
+    MCP_PORT = get_env_variable("MCP_PORT")
+
+    MCP_URL = f"http://{MCP_SERVER}:{MCP_PORT}/mcp"
+
+    DEVICE_NAME = get_env_variable("DEVICE_NAME_FOR_SET_USER_PASSWORD")    
+
+    # Read value from .env
+    DEVICE_NAME = get_env_variable("DEVICE_NAME_FOR_SET_USER_PASSWORD")
+    PLAIN_PASSWORD = get_env_variable("PLAIN_TEXT_PASSWORD_FOR_SET_USER_PASSWORD")
+
+    NEW_USER_PASSWORD_HASH = generate_junos_hash(PLAIN_PASSWORD)
+
+    # Config snippet to set the encrypted password
+    CONFIG_SNIPPET = f"""
+    set system login user guardx authentication encrypted-password "{NEW_USER_PASSWORD_HASH}"
+    """
+
+    # Connect to the MCP server via streamable HTTP
+    async with streamablehttp_client(MCP_URL) as (read_stream, write_stream, _):
+        async with ClientSession(read_stream, write_stream) as session:
+            await session.initialize()
+
+            print(f"Applying user password to {DEVICE_NAME}...")
+            result = await session.call_tool(
+                name="load_and_commit_config",
+                arguments={
+                    "router_name": DEVICE_NAME,
+                    "config_text": CONFIG_SNIPPET.strip(),
+                    "format": "set",
+                    "commit": True,
+                    "commit_comment": "User password updated via MCP"
+                }
+            )
+
+            if result.isError:
+                print(f"Failed to update user password:")
+                for c in result.content:
+                    print(c.text)
+            else:
+                print("User password updated successfully")
+
+if __name__ == "__main__":
+    asyncio.run(main())