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 

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

8 

9""" 

10 

11# Standard 

12from copy import deepcopy 

13 

14toolops_spec_template = { 

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

16 "description": None, 

17 "display_name": None, 

18 "id": None, 

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

20 "is_async": False, 

21 "name": None, 

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

23 "permission": "read_only", 

24} 

25 

26 

27def convert_to_toolops_spec(mcp_cf_tool): 

28 """ 

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

30 

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

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

33 the general structure similar to the MCP tool format. 

34 

35 Args: 

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

37 fields: 'description', 'displayName', 'id', 

38 'inputSchema', and optionally 'outputSchema'. 

39 

40 Returns: 

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

42 - description 

43 - display_name 

44 - id 

45 - input_schema (description, properties, required) 

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

47 - name 

48 

49 Example: 

50 >>> mcp_cf_tool = { 

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

52 ... "displayName": "Sample Tool", 

53 ... "id": "tool123", 

54 ... "inputSchema": { 

55 ... "description": "Input data", 

56 ... "properties": {}, 

57 ... "required": [] 

58 ... }, 

59 ... "outputSchema": { 

60 ... "description": "Output data", 

61 ... "properties": {}, 

62 ... "required": [] 

63 ... }, 

64 ... "name": "Sample Tool Name" 

65 ... } 

66 >>> convert_to_toolops_spec(mcp_cf_tool) 

67 {'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'} 

68 """ 

69 

70 toolops_spec = deepcopy(toolops_spec_template) 

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

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

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

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

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

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

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

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

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

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

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

82 else: 

83 toolops_spec["output_schema"] = {} 

84 return toolops_spec 

85 

86 

87def post_process_nl_test_cases(nl_test_cases): 

88 """ 

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

90 

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

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

93 

94 Args: 

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

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

97 

98 Returns: 

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

100 

101 Example: 

102 >>> nl_test_cases = { 

103 ... "Test_scenarios": [ 

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

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

106 ... ] 

107 ... } 

108 >>> post_process_nl_test_cases(nl_test_cases) 

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

110 

111 """ 

112 test_cases = nl_test_cases.get("Test_scenarios") 

113 for tc in test_cases: 

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

115 del tc[un_wanted] 

116 return test_cases 

117 

118 

119# if __name__ == "__main__": 

120# import json 

121# import os 

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

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

124# for mcp_cf_tool in mcp_cf_tools: 

125# toolops_spec = convert_to_toolops_spec(mcp_cf_tool) 

126# print(toolops_spec)