MCP Test MCP

Instructions

Implementation Reference

• node-wrapper/python-src/src/mcp_test_mcp/tools/tools.py:143-313 (handler)

  The core handler implementation for the 'call_tool' tool. This function is decorated with @mcp.tool for automatic registration. It verifies an active connection, calls the target tool via the MCP client, extracts and formats the result, handles multiple error cases (connection errors, tool not found, invalid args, execution failures), updates connection statistics, provides progress updates via ctx, and returns a comprehensive structured response dictionary.
  @mcp.tool async def call_tool( name: Annotated[str, "Name of the tool to execute on the target MCP server"], arguments: Annotated[dict[str, Any], "Dictionary of arguments to pass to the tool"], ctx: Context ) -> 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. 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() # User-facing progress update await ctx.info(f"Calling tool '{name}' on target server") # Detailed technical log 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) # User-facing success update await ctx.info(f"Tool '{name}' executed successfully") # Detailed technical log 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 # User-facing error update await ctx.error(f"Not connected when calling tool '{name}': {str(e)}") # Detailed technical log 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}'" # User-facing error update await ctx.error(f"Failed to call tool '{name}': {str(e)}") # Detailed technical log 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), }, }
• node-wrapper/python-src/src/mcp_test_mcp/server.py:94-99 (registration)

  Server module imports the shared FastMCP instance (mcp) and explicitly imports tool modules including 'tools' (which contains call_tool). This triggers the @mcp.tool decorator execution, automatically registering the 'call_tool' tool with the MCP server instance.
  from .mcp_instance import mcp # Import tool modules to trigger decorator registration # This MUST happen after mcp instance is imported but before the server runs from .tools import connection, tools, resources, prompts, llm
• node-wrapper/python-src/src/mcp_test_mcp/mcp_instance.py:10-11 (registration)

  Creates the shared FastMCP server instance named 'mcp-test-mcp' that all @mcp.tool decorators register tools with, including the 'call_tool' tool.
  # Create the shared FastMCP server instance mcp = FastMCP(name="mcp-test-mcp")