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

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 typing import Any, Dict, Optional 

10 

11from agentos.mcp import MCPClient, MCPToolInfo 

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

13 

14 

15class MCPToolAdapter(BaseTool): 

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

17 

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

19 handling execution, schema export, and permission routing. 

20 

21 Usage: 

22 adapter = MCPToolAdapter( 

23 client=mcp_client, 

24 tool_info=tool_info, 

25 permission_level=PermissionLevel.MODERATE, 

26 ) 

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

28 """ 

29 

30 def __init__( 

31 self, 

32 client: MCPClient, 

33 tool_info: MCPToolInfo, 

34 permission_level: PermissionLevel = PermissionLevel.MODERATE, 

35 tool_id: Optional[str] = None, 

36 ): 

37 """Initialize the adapter. 

38 

39 Args: 

40 client: Connected MCPClient instance. 

41 tool_info: Tool metadata from MCP discovery. 

42 permission_level: AgentOS permission level for this tool. 

43 tool_id: Optional unique tool identifier. 

44 """ 

45 self._client = client 

46 self._tool_info = tool_info 

47 self._server_name = tool_info.server_name 

48 self._tool_name = tool_info.name 

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

50 self.permission_level = permission_level 

51 

52 @property 

53 def name(self) -> str: 

54 return self._id 

55 

56 @property 

57 def description(self) -> str: 

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

59 

60 def parameters(self) -> dict: 

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

62 return self._tool_info.input_schema 

63 

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

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

66 try: 

67 result = await self._client.call_tool( 

68 self._server_name, 

69 self._tool_name, 

70 arguments, 

71 ) 

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

73 except Exception as e: 

74 return ToolResult.fail( 

75 call_id=str(id(arguments)), 

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

77 ) 

78 

79 def to_openai_schema(self) -> dict: 

80 params = self._tool_info.input_schema 

81 return { 

82 "type": "function", 

83 "function": { 

84 "name": self._id, 

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

86 "parameters": { 

87 **params, 

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

89 } if params else {"type": "object", "properties": {}}, 

90 }, 

91 } 

92 

93 def to_anthropic_schema(self) -> dict: 

94 return { 

95 "name": self._id, 

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

97 "input_schema": self._tool_info.input_schema or { 

98 "type": "object", 

99 "properties": {}, 

100 }, 

101 } 

102 

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

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

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

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

107 name_lower = self._tool_name.lower() 

108 for kw in write_keywords: 

109 if kw in name_lower: 

110 return True 

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

112 

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

114 return not self.is_write_operation(arguments) 

115 

116 def extract_target_path(self, arguments: dict) -> Optional[str]: 

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

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

119 if key in arguments: 

120 return arguments[key] 

121 return None 

122 

123 

124class MCPToolRegistry: 

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

126 

127 Creates MCPToolAdapter wrappers for each discovered tool, with 

128 appropriate permission level assignment. 

129 

130 Usage: 

131 registry = MCPToolRegistry(client) 

132 tools = registry.get_all_tools() 

133 # tools can now be used with any AgentOS agent 

134 """ 

135 

136 def __init__( 

137 self, 

138 client: MCPClient, 

139 default_permission: PermissionLevel = PermissionLevel.MODERATE, 

140 ): 

141 """Initialize the registry. 

142 

143 Args: 

144 client: Connected MCPClient with discovered tools. 

145 default_permission: Default permission level for adapted tools. 

146 """ 

147 self._client = client 

148 self._default_permission = default_permission 

149 self._adapters: Dict[str, MCPToolAdapter] = {} 

150 self._build_adapters() 

151 

152 def _build_adapters(self) -> None: 

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

154 self._adapters.clear() 

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

156 adapter = MCPToolAdapter( 

157 client=self._client, 

158 tool_info=tool_info, 

159 permission_level=self._default_permission, 

160 ) 

161 self._adapters[adapter.name] = adapter 

162 

163 def get_all_tools(self) -> Dict[str, BaseTool]: 

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

165 return dict(self._adapters) 

166 

167 def get_tool(self, name: str) -> Optional[BaseTool]: 

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

169 return self._adapters.get(name) 

170 

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

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

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

174 

175 def refresh(self) -> None: 

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

177 self._build_adapters()