72 lines
2.5 KiB
Python
72 lines
2.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Test script for the GhydraMCP bridge using the MCP client.
|
|
This script tests the bridge by sending MCP requests and handling responses.
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
from typing import Any
|
|
|
|
import anyio
|
|
|
|
from mcp.client.session import ClientSession
|
|
from mcp.client.stdio import StdioServerParameters, stdio_client
|
|
|
|
# Set up logging
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("mcp_client_test")
|
|
|
|
async def test_bridge():
|
|
"""Test the bridge using the MCP client"""
|
|
# Configure the server parameters
|
|
server_parameters = StdioServerParameters(
|
|
command=sys.executable,
|
|
args=["bridge_mcp_hydra.py"],
|
|
)
|
|
|
|
# Connect to the bridge
|
|
logger.info("Connecting to bridge...")
|
|
async with stdio_client(server_parameters) as (read_stream, write_stream):
|
|
# Create a session
|
|
logger.info("Creating session...")
|
|
async with ClientSession(read_stream, write_stream) as session:
|
|
# Initialize the session
|
|
logger.info("Initializing session...")
|
|
init_result = await session.initialize()
|
|
logger.info(f"Initialization result: {init_result}")
|
|
|
|
# List tools
|
|
logger.info("Listing tools...")
|
|
tools_result = await session.list_tools()
|
|
logger.info(f"Tools result: {tools_result}")
|
|
|
|
# Call the list_instances tool
|
|
logger.info("Calling list_instances tool...")
|
|
list_instances_result = await session.call_tool("list_instances")
|
|
logger.info(f"List instances result: {list_instances_result}")
|
|
|
|
# Call the discover_instances tool
|
|
logger.info("Calling discover_instances tool...")
|
|
discover_instances_result = await session.call_tool("discover_instances")
|
|
logger.info(f"Discover instances result: {discover_instances_result}")
|
|
|
|
# Call the list_functions tool
|
|
logger.info("Calling list_functions tool...")
|
|
list_functions_result = await session.call_tool(
|
|
"list_functions",
|
|
arguments={"port": 8192, "offset": 0, "limit": 5}
|
|
)
|
|
logger.info(f"List functions result: {list_functions_result}")
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
try:
|
|
anyio.run(test_bridge)
|
|
except Exception as e:
|
|
logger.error(f"Error: {e}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|