"""Tool testing tools for MCP server testing.
This module provides MCP tools for discovering and executing tools on connected
target MCP servers, enabling comprehensive tool testing workflows.
"""
import logging
import time
from typing import Any
from ..connection import ConnectionError, ConnectionManager
logger = logging.getLogger(__name__)
async def list_tools() -> dict[str, Any]:
"""List all tools available on the connected MCP server.
Retrieves comprehensive information about all tools exposed by the target
server, including full input schemas to enable accurate tool invocation.
Returns:
Dictionary with tool listing including:
- success: True on successful retrieval
- tools: List of tool objects with name, description, and full input_schema
- metadata: Total count, server info, timing information
Raises:
Returns error dict if not connected or retrieval fails
"""
start_time = time.perf_counter()
try:
# Verify connection exists
client, state = ConnectionManager.require_connection()
logger.info("Listing tools from connected MCP server")
# Get tools from the server
tools_result = await client.list_tools()
elapsed_ms = (time.perf_counter() - start_time) * 1000
# Convert tools to dictionary format with full schemas
# Note: client.list_tools() returns a list directly, not an object with .tools
tools_list = []
for tool in tools_result:
# inputSchema is already a dict, not a Pydantic model
input_schema = tool.inputSchema if hasattr(tool, "inputSchema") and tool.inputSchema else {}
tool_dict = {
"name": tool.name,
"description": tool.description if tool.description else "",
"input_schema": input_schema,
}
tools_list.append(tool_dict)
metadata = {
"total_tools": len(tools_list),
"server_url": state.server_url,
"retrieved_at": time.time(),
"request_time_ms": round(elapsed_ms, 2),
}
# Add server info if available
if state.server_info:
metadata["server_name"] = state.server_info.get("name", "unknown")
metadata["server_version"] = state.server_info.get("version")
logger.info(
f"Retrieved {len(tools_list)} tools from server",
extra={
"tool_count": len(tools_list),
"server_url": state.server_url,
"duration_ms": elapsed_ms,
},
)
return {
"success": True,
"tools": tools_list,
"metadata": metadata,
}
except ConnectionError as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
logger.error(f"Not connected: {str(e)}", extra={"duration_ms": elapsed_ms})
return {
"success": False,
"error": {
"error_type": "not_connected",
"message": str(e),
"details": {},
"suggestion": "Use connect_to_server() to establish a connection first",
},
"tools": [],
"metadata": {
"request_time_ms": round(elapsed_ms, 2),
},
}
except Exception as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
logger.exception("Failed to list tools", extra={"duration_ms": elapsed_ms})
# Increment error counter
ConnectionManager.increment_stat("errors")
return {
"success": False,
"error": {
"error_type": "execution_error",
"message": f"Failed to list tools: {str(e)}",
"details": {"exception_type": type(e).__name__},
"suggestion": "Check that the server supports the tools capability and is responding correctly",
},
"tools": [],
"metadata": {
"request_time_ms": round(elapsed_ms, 2),
},
}
async def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
"""Execute a tool on the connected MCP server.
Calls a tool by name with the provided arguments and returns the result
along with execution timing and metadata.
Args:
name: Name of the tool to execute
arguments: Dictionary of arguments to pass to the tool
Returns:
Dictionary with tool execution results including:
- success: True if tool executed successfully
- tool_call: Object with tool_name, arguments, result, and execution metadata
- metadata: Request timing and server information
Raises:
Returns error dict for various failure scenarios:
- not_connected: No active connection
- tool_not_found: Tool doesn't exist on server
- invalid_arguments: Arguments don't match tool schema
- execution_error: Tool execution failed
"""
start_time = time.perf_counter()
try:
# Verify connection exists
client, state = ConnectionManager.require_connection()
logger.info(
f"Calling tool '{name}' with arguments",
extra={"tool_name": name, "arguments": arguments},
)
# Execute the tool
tool_start = time.perf_counter()
result = await client.call_tool(name, arguments)
tool_elapsed_ms = (time.perf_counter() - tool_start) * 1000
# Increment statistics
ConnectionManager.increment_stat("tools_called")
total_elapsed_ms = (time.perf_counter() - start_time) * 1000
# Extract result content
result_content = None
if hasattr(result, "content") and result.content:
# Handle list of content items
if isinstance(result.content, list) and len(result.content) > 0:
content_item = result.content[0]
if hasattr(content_item, "text"):
result_content = content_item.text
elif hasattr(content_item, "data"):
result_content = content_item.data
else:
result_content = str(content_item)
else:
result_content = result.content
elif hasattr(result, "result"):
result_content = result.result
else:
result_content = str(result)
logger.info(
f"Tool '{name}' executed successfully",
extra={
"tool_name": name,
"execution_ms": tool_elapsed_ms,
"total_ms": total_elapsed_ms,
},
)
return {
"success": True,
"tool_call": {
"tool_name": name,
"arguments": arguments,
"result": result_content,
"execution": {
"duration_ms": round(tool_elapsed_ms, 2),
"success": True,
},
},
"metadata": {
"request_time_ms": round(total_elapsed_ms, 2),
"server_url": state.server_url,
"connection_statistics": state.statistics,
},
}
except ConnectionError as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
logger.error(
f"Not connected when calling tool '{name}': {str(e)}",
extra={"tool_name": name, "duration_ms": elapsed_ms},
)
return {
"success": False,
"error": {
"error_type": "not_connected",
"message": str(e),
"details": {"tool_name": name},
"suggestion": "Use connect_to_server() to establish a connection first",
},
"tool_call": None,
"metadata": {
"request_time_ms": round(elapsed_ms, 2),
},
}
except Exception as e:
elapsed_ms = (time.perf_counter() - start_time) * 1000
# Determine error type based on exception message
error_type = "execution_error"
suggestion = "Check the tool name and arguments, then retry"
error_msg = str(e).lower()
if "not found" in error_msg or "unknown tool" in error_msg:
error_type = "tool_not_found"
suggestion = f"Tool '{name}' does not exist on the server. Use list_tools() to see available tools"
elif "argument" in error_msg or "parameter" in error_msg or "validation" in error_msg:
error_type = "invalid_arguments"
suggestion = f"Arguments do not match the tool schema. Use list_tools() to see the correct schema for '{name}'"
logger.error(
f"Failed to call tool '{name}': {str(e)}",
extra={
"tool_name": name,
"arguments": arguments,
"error_type": error_type,
"duration_ms": elapsed_ms,
},
)
# Increment error counter
ConnectionManager.increment_stat("errors")
return {
"success": False,
"error": {
"error_type": error_type,
"message": f"Failed to call tool '{name}': {str(e)}",
"details": {
"tool_name": name,
"arguments": arguments,
"exception_type": type(e).__name__,
},
"suggestion": suggestion,
},
"tool_call": None,
"metadata": {
"request_time_ms": round(elapsed_ms, 2),
},
}
