Coverage for mcpgateway / routers / toolops_router.py: 96%

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

2"""Location: ./mcpgateway/routers/toolops_router.py 

3Copyright 2025 

4SPDX-License-Identifier: Apache-2.0 

5Authors: Jay Bandlamudi 

6 

7Toolops Router Module 

8 

9This module provides FastAPI endpoints for managing Toolops functionalities 

10.It supports tool test case generation , tool meta-data enrichment and tool 

11test case execution with an agent. 

12 

13The module handles API endpoints created for several toolops features. 

14 

15""" 

16 

17# Standard 

18from typing import Any, Dict, List 

19 

20# Third-Party 

21from fastapi import APIRouter, Depends, HTTPException, Query, status 

22import orjson 

23from pydantic import BaseModel, Field 

24from sqlalchemy.orm import Session 

25 

26# First-Party 

27from mcpgateway.main import get_db 

28from mcpgateway.middleware.rbac import get_current_user_with_permissions, require_permission 

29from mcpgateway.services.logging_service import LoggingService 

30from mcpgateway.services.tool_service import ToolService 

31from mcpgateway.toolops.toolops_altk_service import enrich_tool, execute_tool_nl_test_cases, validation_generate_test_cases 

32 

33# Initialize router 

34toolops_router = APIRouter(prefix="/toolops", tags=["Toolops"]) 

35 

36# Initialize services 

37tool_service = ToolService() 

38 

39# Logging 

40logging_service = LoggingService() 

41logger = logging_service.get_logger(__name__) 

42 

43# ---------- Utility ---------- 

44 

45 

46class ToolNLTestInput(BaseModel): 

47 """ 

48 Toolops test input format to run NL test cases of a tool using agent 

49 

50 Args: 

51 tool_id : Unique Tool ID 

52 tool_nl_test_cases: List of natural language test cases for testing MCP tool with the agent 

53 

54 Returns: 

55 This class defines tool NL test input format and returns nothing. 

56 """ 

57 

58 tool_id: str | None = Field(default=None, title="Tool ID", max_length=300) 

59 tool_nl_test_cases: list | None = Field(default=None, title="List of natural language test cases for testing MCP tool with the agent") 

60 

61 

62# ---------- ROUTES ---------- 

63 

64 

65# First-Party 

66# Toolops APIs - Generating test cases , Tool enrichment # 

67@toolops_router.post("/validation/generate_testcases") 

68@require_permission("admin.system_config") 

69async def generate_testcases_for_tool( 

70 tool_id: str = Query(None, description="Tool ID"), 

71 number_of_test_cases: int = Query(2, description="Maximum number of tool test cases"), 

72 number_of_nl_variations: int = Query(1, description="Number of NL utterance variations per test case"), 

73 mode: str = Query("generate", description="Three modes: 'generate' for test case generation, 'query' for obtaining test cases from DB , 'status' to check test generation status"), 

74 db: Session = Depends(get_db), 

75 _user=Depends(get_current_user_with_permissions), 

76) -> List[Dict]: 

77 """ 

78 Generate test cases for a tool 

79 

80 This endpoint handles the automated test case generation for a tool by accepting 

81 a tool id . The `require_auth` dependency ensures that 

82 the user is authenticated before proceeding. 

83 

84 Args: 

85 tool_id: Tool ID in context forge. 

86 number_of_test_cases: Number of test cases to generate for the given tools (optional) 

87 number_of_nl_variations: Number of Natural language variations(parapharses) per test case (optional) 

88 mode: Three supported modes - 'generate' for test case generation, 'query' for obtaining test cases from DB , 'status' to check test generation status 

89 db: DB session to connect with database 

90 

91 Returns: 

92 List: A list of test cases generated for the tool , each test case is dictionary object 

93 

94 Raises: 

95 HTTPException: If the request body contains invalid JSON, a 400 Bad Request error is raised. 

96 """ 

97 try: 

98 # logger.debug(f"Authenticated user {user} is initializing the protocol.") 

99 test_cases = await validation_generate_test_cases(tool_id, tool_service, db, number_of_test_cases, number_of_nl_variations, mode) 

100 return test_cases 

101 

102 except orjson.JSONDecodeError: 

103 raise HTTPException( 

104 status_code=status.HTTP_400_BAD_REQUEST, 

105 detail="Invalid JSON in request body", 

106 ) 

107 

108 

109@toolops_router.post("/validation/execute_tool_nl_testcases") 

110@require_permission("admin.system_config") 

111async def execute_tool_nl_testcases(tool_nl_test_input: ToolNLTestInput, db: Session = Depends(get_db), _user=Depends(get_current_user_with_permissions)) -> List: 

112 """ 

113 Execute test cases for a tool 

114 

115 This endpoint handles the automated test case generation for a tool by accepting 

116 a tool id . The `require_auth` dependency ensures that 

117 the user is authenticated before proceeding. 

118 

119 Args: 

120 tool_nl_test_input: NL test case format input to run test cases with agent , it contains\ 

121 - tool_id: Tool ID in context forge\ 

122 - tool_nl_test_cases: List of natural language test cases (utteances) for testing MCP tool with the agent 

123 db: DB session to connect with database 

124 

125 Returns: 

126 List: A list of tool outputs after agent execution for the provided tool nl test cases 

127 

128 Raises: 

129 HTTPException: If the request body contains invalid JSON, a 400 Bad Request error is raised. 

130 """ 

131 try: 

132 # logger.debug(f"Authenticated user {user} is initializing the protocol.") 

133 tool_id = tool_nl_test_input.tool_id 

134 tool_nl_test_cases = tool_nl_test_input.tool_nl_test_cases 

135 tool_nl_test_cases_output = await execute_tool_nl_test_cases(tool_id, tool_nl_test_cases, tool_service, db) 

136 return tool_nl_test_cases_output 

137 

138 except orjson.JSONDecodeError: 

139 raise HTTPException( 

140 status_code=status.HTTP_400_BAD_REQUEST, 

141 detail="Invalid JSON in request body", 

142 ) 

143 

144 

145@toolops_router.post("/enrichment/enrich_tool") 

146@require_permission("admin.system_config") 

147async def enrich_a_tool(tool_id: str = Query(None, description="Tool ID"), db: Session = Depends(get_db), _user=Depends(get_current_user_with_permissions)) -> dict[str, Any]: 

148 """ 

149 Enriches an input tool 

150 

151 Args: 

152 tool_id: Unique Tool ID MCP-CF. 

153 db: The database session used to interact with the data store. 

154 

155 Returns: 

156 result: A dict having the keys "tool_id", "tool_name", "original_desc" and "enriched_desc" with their corresponding values 

157 

158 Raises: 

159 HTTPException: If the request body contains invalid JSON, a 400 Bad Request error is raised. 

160 """ 

161 try: 

162 logger.info("Running tool enrichment for Tool - " + tool_id) 

163 enriched_tool_description, tool_schema = await enrich_tool(tool_id, tool_service, db) 

164 result: dict[str, Any] = {} 

165 result["tool_id"] = tool_id 

166 result["tool_name"] = tool_schema.name 

167 result["original_desc"] = tool_schema.description 

168 result["enriched_desc"] = enriched_tool_description 

169 # logger.info ("result: "+ json.dumps(result, indent=4, sort_keys=False)) 

170 return result 

171 

172 except Exception as e: 

173 logger.info("Error in tool enrichment for Tool - " + str(e)) 

174 raise HTTPException( 

175 status_code=status.HTTP_400_BAD_REQUEST, 

176 detail="Invalid JSON in request body" + str(e), 

177 ) from e