Coverage for src / mcp_server_langgraph / tools / tool_discovery.py: 98%

1""" 

2Tool discovery for progressive disclosure 

3 

4Implements search_tools endpoint following Anthropic's MCP best practices 

5for token-efficient tool discovery. 

6""" 

7 

8import json 

9import logging 

10from typing import Literal 

11 

12from langchain_core.tools import BaseTool, tool 

13from pydantic import BaseModel, Field 

14 

15from mcp_server_langgraph.tools import ALL_TOOLS 

16 

17logger = logging.getLogger(__name__) 

18 

19DetailLevel = Literal["minimal", "standard", "full"] 

20 

21 

22class SearchToolsInput(BaseModel): 

23 """Input schema for search_tools""" 

24 

25 query: str | None = Field(default=None, description="Search query (keyword or category)") 

26 category: str | None = Field( 

27 default=None, 

28 description="Tool category (calculator, search, filesystem, execution)", 

29 ) 

30 detail_level: DetailLevel = Field( 

31 default="minimal", 

32 description="Level of detail: minimal (name+desc), standard (+params), full (+examples)", 

33 ) 

34 

35 

36def _filter_tools_by_category(category: str) -> list[BaseTool]: 

37 """Filter tools by category.""" 

38 category_lower = category.lower() 

39 if category_lower == "calculator": 

40 from mcp_server_langgraph.tools import CALCULATOR_TOOLS 

41 

42 return CALCULATOR_TOOLS 

43 elif category_lower == "search": 

44 from mcp_server_langgraph.tools import SEARCH_TOOLS 

45 

46 return SEARCH_TOOLS 

47 elif category_lower == "filesystem": 

48 from mcp_server_langgraph.tools import FILESYSTEM_TOOLS 

49 

50 return FILESYSTEM_TOOLS 

51 elif category_lower == "execution": 

52 from mcp_server_langgraph.tools import CODE_EXECUTION_TOOLS 

53 

54 return CODE_EXECUTION_TOOLS 

55 return ALL_TOOLS 

56 

57 

58def _filter_tools_by_query(tools: list[BaseTool], query: str) -> list[BaseTool]: 

59 """Filter tools by search query.""" 

60 query_lower = query.lower() 

61 return [t for t in tools if query_lower in t.name.lower() or query_lower in (t.description or "").lower()] 

62 

63 

64def _format_tool_minimal(t: BaseTool) -> str: 

65 """Format tool in minimal mode.""" 

66 return f"- **{t.name}**: {t.description}\n" 

67 

68 

69def _format_tool_standard(t: BaseTool) -> str: 

70 """Format tool in standard mode.""" 

71 result = f"### {t.name}\n{t.description}\n\n" 

72 

73 if hasattr(t, "args_schema") and t.args_schema: 

74 schema = t.args_schema.model_json_schema() # type: ignore[union-attr] 

75 if "properties" in schema: 75 ↛ 81line 75 didn't jump to line 81 because the condition on line 75 was always true

76 result += "**Parameters:**\n" 

77 for param_name, param_info in schema["properties"].items(): 

78 param_desc = param_info.get("description", "No description") 

79 param_type = param_info.get("type", "any") 

80 result += f"- `{param_name}` ({param_type}): {param_desc}\n" 

81 result += "\n" 

82 

83 return result 

84 

85 

86def _format_tool_full(t: BaseTool) -> str: 

87 """Format tool in full mode.""" 

88 result = f"### {t.name}\n{t.description}\n\n" 

89 

90 if hasattr(t, "args_schema") and t.args_schema: 

91 schema = t.args_schema.model_json_schema() # type: ignore[union-attr] 

92 if "properties" in schema: 92 ↛ 105line 92 didn't jump to line 105 because the condition on line 92 was always true

93 result += "**Parameters:**\n" 

94 for param_name, param_info in schema["properties"].items(): 

95 param_desc = param_info.get("description", "No description") 

96 param_type = param_info.get("type", "any") 

97 required = param_name in schema.get("required", []) 

98 req_str = "required" if required else "optional" 

99 result += f"- `{param_name}` ({param_type}, {req_str}): {param_desc}\n" 

100 

101 result += "\n**Full Schema:**\n```json\n" 

102 result += json.dumps(schema, indent=2) 

103 result += "\n```\n\n" 

104 

105 return result 

106 

107 

108def _format_tool_results(tools: list[BaseTool], detail_level: str) -> str: 

109 """Format tool results based on detail level.""" 

110 result = f"Found {len(tools)} tool(s):\n\n" 

111 

112 if detail_level == "minimal": 

113 for t in tools: 

114 result += _format_tool_minimal(t) 

115 elif detail_level == "standard": 

116 for t in tools: 

117 result += _format_tool_standard(t) 

118 else: # full 

119 for t in tools: 

120 result += _format_tool_full(t) 

121 

122 return result 

123 

124 

125@tool 

126def search_tools( 

127 query: str | None = None, 

128 category: str | None = None, 

129 detail_level: str = "minimal", 

130) -> str: 

131 """ 

132 Search and discover available tools (progressive disclosure for token efficiency). 

133 

134 Implements Anthropic's best practice for progressive tool discovery, allowing 

135 agents to query tools by keyword or category rather than loading all tool 

136 definitions upfront. This can save 98%+ tokens compared to listing all tools. 

137 

138 Args: 

139 query: Optional keyword search (searches name and description) 

140 category: Optional category filter (calculator, search, filesystem, execution) 

141 detail_level: Level of detail - minimal, standard, or full 

142 

143 Returns: 

144 Formatted list of matching tools with requested detail level 

145 

146 Example: 

147 >>> search_tools.invoke({"category": "calculator", "detail_level": "minimal"}) 

148 "Found 5 tools:\\n- calculator: Evaluate mathematical expressions\\n..." 

149 """ 

150 # Get all tools or filter by category 

151 tools = _filter_tools_by_category(category) if category else ALL_TOOLS 

152 

153 # Filter by query if provided 

154 if query: 

155 tools = _filter_tools_by_query(tools, query) 

156 

157 if not tools: 

158 return f"No tools found matching criteria (query={query}, category={category})" 

159 

160 # Format and return results 

161 return _format_tool_results(tools, detail_level)