Coverage for merco/mcp/tool.py: 100%

1"""MCP tool → BaseTool adapter for ToolRegistry.""" 

2from merco.tools.base import BaseTool 

3 

4 

5class MCPServerTool(BaseTool): 

6 """Adapt an MCP tool spec to a merco ToolRegistry-compatible tool.""" 

7 

8 def __init__(self, mcp_spec: dict, server_name: str, handler, guard=None): 

9 self.name = mcp_spec.get("name", "unknown") 

10 self.description = mcp_spec.get("description", f"MCP tool from {server_name}") 

11 self.toolset = f"mcp:{server_name}" 

12 self.server = server_name 

13 self._input_schema = mcp_spec.get("inputSchema", {}) 

14 self._handler = handler # async (tool_name, arguments) -> dict 

15 self._guard = guard 

16 

17 @property 

18 def parameters(self) -> dict: 

19 return self._input_schema 

20 

21 async def execute(self, **kwargs) -> dict: 

22 # ToolGuard check — sandbox integration 

23 if self._guard: 

24 approved = await self._guard.check( 

25 self.name, kwargs, source=f"mcp:{self.server}" 

26 ) 

27 if not approved: 

28 return {"error": "操作已被拦截或取消"} 

29 try: 

30 return await self._handler(self.name, kwargs) 

31 except Exception as e: 

32 return {"error": f"{self.name}: {e}", "isError": True} 

33 

34 def check(self) -> bool: 

35 return self._handler is not None 

36 

37 def get_definition(self, context=None) -> dict: 

38 return { 

39 "type": "function", 

40 "function": { 

41 "name": self.name, 

42 "description": self.description, 

43 "parameters": self._input_schema, 

44 }, 

45 }