Coverage for mcpgateway / toolops / utils / format_conversion.py: 100%

1# -*- coding: utf-8 -*- 

2"""Location: ./mcpgateway/toolops/utils/tool_format_conversion.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Jay Bandlamudi 

6 

7MCP Gateway - module for converting MCP tool format to toolops specific internal format. 

8 

9""" 

10# Standard 

11from copy import deepcopy 

12 

13toolops_spec_template = { 

14 "binding": {"python": {"connections": {}, "function": "mcp-cf-tool-default", "requirements": []}}, 

15 "description": None, 

16 "display_name": None, 

17 "id": None, 

18 "input_schema": {"description": None, "properties": {}, "required": [], "type": "object"}, 

19 "is_async": False, 

20 "name": None, 

21 "output_schema": {"description": None, "properties": {}, "required": [], "type": "object"}, 

22 "permission": "read_only", 

23} 

24 

25 

26def convert_to_toolops_spec(mcp_cf_tool): 

27 """ 

28 Converts an MCP tool JSON format to ToolOps-specific internal format. 

29 

30 This method takes a JSON representation of an MCP tool and converts it into a 

31 format that is compatible with ToolOps' internal specification, while keeping 

32 the general structure similar to the MCP tool format. 

33 

34 Args: 

35 mcp_cf_tool (dict): The MCP tool in JSON format. It must contain the 

36 fields: 'description', 'displayName', 'id', 

37 'inputSchema', and optionally 'outputSchema'. 

38 

39 Returns: 

40 dict: The ToolOps-specific internal format of the tool, which includes: 

41 - description 

42 - display_name 

43 - id 

44 - input_schema (description, properties, required) 

45 - output_schema (description, properties, required, or an empty dict if not present) 

46 - name 

47 

48 Example: 

49 >>> mcp_cf_tool = { 

50 ... "description": "A sample tool", 

51 ... "displayName": "Sample Tool", 

52 ... "id": "tool123", 

53 ... "inputSchema": { 

54 ... "description": "Input data", 

55 ... "properties": {}, 

56 ... "required": [] 

57 ... }, 

58 ... "outputSchema": { 

59 ... "description": "Output data", 

60 ... "properties": {}, 

61 ... "required": [] 

62 ... }, 

63 ... "name": "Sample Tool Name" 

64 ... } 

65 >>> convert_to_toolops_spec(mcp_cf_tool) 

66 {'binding': {'python': {'connections': {}, 'function': 'mcp-cf-tool-default', 'requirements': []}}, 'description': 'A sample tool', 'display_name': 'Sample Tool', 'id': 'tool123', 'input_schema': {'description': 'Input data', 'properties': {}, 'required': [], 'type': 'object'}, 'is_async': False, 'name': 'Sample Tool Name', 'output_schema': {'description': 'Output data', 'properties': {}, 'required': [], 'type': 'object'}, 'permission': 'read_only'} 

67 """ 

68 

69 toolops_spec = deepcopy(toolops_spec_template) 

70 toolops_spec["description"] = mcp_cf_tool.get("description", None) 

71 toolops_spec["display_name"] = mcp_cf_tool.get("displayName", None) 

72 toolops_spec["id"] = mcp_cf_tool.get("id", None) 

73 toolops_spec["input_schema"]["description"] = mcp_cf_tool.get("inputSchema", {}).get("description", None) 

74 toolops_spec["input_schema"]["properties"] = mcp_cf_tool.get("inputSchema", {}).get("properties", {}) 

75 toolops_spec["input_schema"]["required"] = mcp_cf_tool.get("inputSchema", {}).get("required", []) 

76 toolops_spec["name"] = mcp_cf_tool.get("name", None) 

77 if mcp_cf_tool.get("outputSchema") is not None: 

78 toolops_spec["output_schema"]["description"] = mcp_cf_tool.get("outputSchema", {}).get("description", None) 

79 toolops_spec["output_schema"]["properties"] = mcp_cf_tool.get("outputSchema", {}).get("properties", {}) 

80 toolops_spec["output_schema"]["required"] = mcp_cf_tool.get("outputSchema", {}).get("required", []) 

81 else: 

82 toolops_spec["output_schema"] = {} 

83 return toolops_spec 

84 

85 

86def post_process_nl_test_cases(nl_test_cases): 

87 """ 

88 Post-processes generated test cases to remove unwanted parameters. 

89 

90 This method processes the test cases by removing unnecessary parameters, such as 

91 "scenario_type" and "input", from the input test case dictionary. 

92 

93 Args: 

94 nl_test_cases (dict): A dictionary containing test cases under the key "Test_scenarios". 

95 Each test case may contain parameters like "scenario_type", "input", etc. 

96 

97 Returns: 

98 list: A list of processed test cases with unwanted parameters removed. 

99 

100 Example: 

101 >>> nl_test_cases = { 

102 ... "Test_scenarios": [ 

103 ... {"scenario_type": "type1", "input": "data1", "other_param": "value1"}, 

104 ... {"scenario_type": "type2", "input": "data2", "other_param": "value2"} 

105 ... ] 

106 ... } 

107 >>> post_process_nl_test_cases(nl_test_cases) 

108 [{'other_param': 'value1'}, {'other_param': 'value2'}] 

109 

110 """ 

111 test_cases = nl_test_cases.get("Test_scenarios") 

112 for tc in test_cases: 

113 for un_wanted in ["scenario_type", "input"]: 

114 del tc[un_wanted] 

115 return test_cases 

116 

117 

118# if __name__ == "__main__": 

119# import json 

120# import os 

121# # mcp_cf_tools = json.load(open('./list_of_tools_from_mcp_cf.json','r')) 

122# mcp_cf_tools = [json.load(open("mcp_cf_spec.json", "r"))] 

123# for mcp_cf_tool in mcp_cf_tools: 

124# toolops_spec = convert_to_toolops_spec(mcp_cf_tool) 

125# print(toolops_spec)