Coverage for agentos/mcp/adapter.py: 37%

1"""MCP Tool Adapter for AgentOS. 

2 

3Wraps MCP tools as AgentOS BaseTool instances, enabling seamless 

4integration with the AgentOS tool system and permission model. 

5""" 

6 

7from __future__ import annotations 

8 

9from agentos.mcp import MCPClient, MCPToolInfo 

10from agentos.tools.base import BaseTool, PermissionLevel, ToolResult 

11 

12 

13class MCPToolAdapter(BaseTool): 

14 """Adapts an MCP tool to the AgentOS BaseTool interface. 

15 

16 Wraps a remote MCP tool call in the standard BaseTool protocol, 

17 handling execution, schema export, and permission routing. 

18 

19 Usage: 

20 adapter = MCPToolAdapter( 

21 client=mcp_client, 

22 tool_info=tool_info, 

23 permission_level=PermissionLevel.MODERATE, 

24 ) 

25 result = await adapter.execute({"path": "/tmp/test.txt"}) 

26 """ 

27 

28 def __init__( 

29 self, 

30 client: MCPClient, 

31 tool_info: MCPToolInfo, 

32 permission_level: PermissionLevel = PermissionLevel.MODERATE, 

33 tool_id: str | None = None, 

34 ): 

35 """Initialize the adapter. 

36 

37 Args: 

38 client: Connected MCPClient instance. 

39 tool_info: Tool metadata from MCP discovery. 

40 permission_level: AgentOS permission level for this tool. 

41 tool_id: Optional unique tool identifier. 

42 """ 

43 self._client = client 

44 self._tool_info = tool_info 

45 self._server_name = tool_info.server_name 

46 self._tool_name = tool_info.name 

47 self._id = tool_id or f"mcp__{self._server_name}__{self._tool_name}" 

48 self.permission_level = permission_level 

49 

50 @property 

51 def name(self) -> str: 

52 return self._id 

53 

54 @property 

55 def description(self) -> str: 

56 return self._tool_info.description or f"MCP tool: {self._tool_name}" 

57 

58 def parameters(self) -> dict: 

59 """Return the JSON Schema for tool parameters.""" 

60 return self._tool_info.input_schema 

61 

62 async def execute(self, arguments: dict, sandbox=None) -> ToolResult: 

63 """Execute the MCP tool and wrap result in ToolResult.""" 

64 try: 

65 result = await self._client.call_tool( 

66 self._server_name, 

67 self._tool_name, 

68 arguments, 

69 ) 

70 return ToolResult.ok(call_id=str(id(arguments)), output=str(result)) 

71 except Exception as e: 

72 return ToolResult.fail( 

73 call_id=str(id(arguments)), 

74 error=f"MCP tool '{self._tool_name}' error: {e}", 

75 ) 

76 

77 def to_openai_schema(self) -> dict: 

78 params = self._tool_info.input_schema 

79 return { 

80 "type": "function", 

81 "function": { 

82 "name": self._id, 

83 "description": self._tool_info.description or "", 

84 "parameters": ( 

85 { 

86 **params, 

87 "title": params.get("title", self._id), 

88 } 

89 if params 

90 else {"type": "object", "properties": {}} 

91 ), 

92 }, 

93 } 

94 

95 def to_anthropic_schema(self) -> dict: 

96 return { 

97 "name": self._id, 

98 "description": self._tool_info.description or "", 

99 "input_schema": self._tool_info.input_schema 

100 or { 

101 "type": "object", 

102 "properties": {}, 

103 }, 

104 } 

105 

106 def is_write_operation(self, arguments: dict) -> bool: 

107 """Heuristic: MCP tools with names containing write/update/create/delete 

108 or having 'mode' parameter are treated as write operations.""" 

109 write_keywords = ("write", "update", "create", "delete", "remove", "put", "patch", "post") 

110 name_lower = self._tool_name.lower() 

111 for kw in write_keywords: 

112 if kw in name_lower: 

113 return True 

114 return arguments.get("mode") == "write" 

115 

116 def is_read_operation(self, arguments: dict) -> bool: 

117 return not self.is_write_operation(arguments) 

118 

119 def extract_target_path(self, arguments: dict) -> str | None: 

120 """Extract file path from common MCP tool arguments.""" 

121 for key in ("path", "uri", "file_path", "filepath"): 

122 if key in arguments: 

123 return arguments[key] 

124 return None 

125 

126 

127class MCPToolRegistry: 

128 """Registry that adapts all tools from an MCPClient into BaseTool instances. 

129 

130 Creates MCPToolAdapter wrappers for each discovered tool, with 

131 appropriate permission level assignment. 

132 

133 Usage: 

134 registry = MCPToolRegistry(client) 

135 tools = registry.get_all_tools() 

136 # tools can now be used with any AgentOS agent 

137 """ 

138 

139 def __init__( 

140 self, 

141 client: MCPClient, 

142 default_permission: PermissionLevel = PermissionLevel.MODERATE, 

143 ): 

144 """Initialize the registry. 

145 

146 Args: 

147 client: Connected MCPClient with discovered tools. 

148 default_permission: Default permission level for adapted tools. 

149 """ 

150 self._client = client 

151 self._default_permission = default_permission 

152 self._adapters: dict[str, MCPToolAdapter] = {} 

153 self._build_adapters() 

154 

155 def _build_adapters(self) -> None: 

156 """Rebuild tool adapters from the current client state.""" 

157 self._adapters.clear() 

158 for tool_info in self._client.list_tools(): 

159 adapter = MCPToolAdapter( 

160 client=self._client, 

161 tool_info=tool_info, 

162 permission_level=self._default_permission, 

163 ) 

164 self._adapters[adapter.name] = adapter 

165 

166 def get_all_tools(self) -> dict[str, BaseTool]: 

167 """Return all adapted tools as name -> BaseTool mapping.""" 

168 return dict(self._adapters) 

169 

170 def get_tool(self, name: str) -> BaseTool | None: 

171 """Get a single adapted tool by name.""" 

172 return self._adapters.get(name) 

173 

174 def get_tool_schemas(self, format: str = "openai") -> list: 

175 """Export schemas for all adapted tools.""" 

176 return [t.to_openai_schema() for t in self._adapters.values()] 

177 

178 def refresh(self) -> None: 

179 """Refresh the registry to pick up new tools.""" 

180 self._build_adapters()