From 6f38122828f94ad3ab8c9195088210993101a961 Mon Sep 17 00:00:00 2001 From: Pradeep H Krishnamurthy Date: Tue, 23 Sep 2025 12:35:19 +0530 Subject: [PATCH] Sample MCP client to execute tools --- test/.env | 10 +++ test/test_mcp_client_gather_device_facts.py | 47 ++++++++++++++ test/test_mcp_client_list_tools.py | 43 +++++++++++++ test/test_mcp_client_set_user_password.py | 70 +++++++++++++++++++++ 4 files changed, 170 insertions(+) create mode 100644 test/.env create mode 100644 test/test_mcp_client_gather_device_facts.py create mode 100644 test/test_mcp_client_list_tools.py create mode 100644 test/test_mcp_client_set_user_password.py diff --git a/test/.env b/test/.env new file mode 100644 index 0000000..5924c97 --- /dev/null +++ b/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= diff --git a/test/test_mcp_client_gather_device_facts.py b/test/test_mcp_client_gather_device_facts.py new file mode 100644 index 0000000..ab743a2 --- /dev/null +++ b/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()) diff --git a/test/test_mcp_client_list_tools.py b/test/test_mcp_client_list_tools.py new file mode 100644 index 0000000..0614883 --- /dev/null +++ b/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()) diff --git a/test/test_mcp_client_set_user_password.py b/test/test_mcp_client_set_user_password.py new file mode 100644 index 0000000..e92449e --- /dev/null +++ b/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())