Coverage for agentos/tests/test_mcp.py: 0%

1"""Tests for MCP client and tool adapter.""" 

2 

3import pytest 

4 

5from agentos.mcp import ( 

6 MCPClient, 

7 MCPError, 

8 MCPPromptInfo, 

9 MCPResourceInfo, 

10 MCPServerConfig, 

11 MCPToolInfo, 

12) 

13from agentos.mcp.adapter import MCPToolAdapter, MCPToolRegistry 

14from agentos.tools.base import PermissionLevel 

15 

16 

17class TestMCPServerConfig: 

18 """Server configuration tests.""" 

19 

20 def test_defaults(self): 

21 config = MCPServerConfig(name="test") 

22 assert config.name == "test" 

23 assert config.transport == "stdio" 

24 assert config.args == [] 

25 assert config.timeout == 30 

26 

27 def test_custom(self): 

28 config = MCPServerConfig( 

29 name="github", 

30 transport="sse", 

31 url="http://localhost:8080", 

32 timeout=60, 

33 ) 

34 assert config.transport == "sse" 

35 assert config.url == "http://localhost:8080" 

36 assert config.timeout == 60 

37 

38 

39class TestMCPClientLifecycle: 

40 """Client init and teardown tests (no real server needed).""" 

41 

42 @pytest.mark.asyncio 

43 async def test_init_empty(self): 

44 client = MCPClient() 

45 assert client.connected_servers == [] 

46 assert client.list_tools() == [] 

47 assert client.list_resources() == [] 

48 assert client.list_prompts() == [] 

49 

50 @pytest.mark.asyncio 

51 async def test_context_manager(self): 

52 async with MCPClient() as client: 

53 assert client.connected_servers == [] 

54 

55 @pytest.mark.asyncio 

56 async def test_connect_unknown_transport(self): 

57 client = MCPClient() 

58 config = MCPServerConfig(name="bad", transport="grpc") 

59 with pytest.raises(MCPError, match="Unknown transport"): 

60 await client.connect_server(config) 

61 

62 @pytest.mark.asyncio 

63 async def test_sse_requires_url(self): 

64 client = MCPClient() 

65 config = MCPServerConfig(name="bad", transport="sse") 

66 with pytest.raises(MCPError, match="URL required"): 

67 await client.connect_server(config) 

68 

69 

70class TestMCPToolAdapter: 

71 """Tool adapter wrapping tests.""" 

72 

73 def test_adapt_tool_basic(self): 

74 client = MCPClient() 

75 tool = MCPToolInfo( 

76 name="read", 

77 description="Read a file", 

78 server_name="fs", 

79 input_schema={ 

80 "type": "object", 

81 "properties": {"path": {"type": "string"}}, 

82 "required": ["path"], 

83 }, 

84 ) 

85 adapter = MCPToolAdapter(client=client, tool_info=tool) 

86 assert adapter.name == "mcp__fs__read" 

87 assert adapter.description == "Read a file" 

88 assert "path" in adapter.parameters()["properties"] 

89 

90 def test_to_openai_schema(self): 

91 client = MCPClient() 

92 tool = MCPToolInfo( 

93 name="search", 

94 description="Search docs", 

95 server_name="docs", 

96 input_schema={"type": "object", "properties": {"q": {"type": "string"}}}, 

97 ) 

98 adapter = MCPToolAdapter(client=client, tool_info=tool) 

99 schema = adapter.to_openai_schema() 

100 assert schema["type"] == "function" 

101 assert schema["function"]["name"] == "mcp__docs__search" 

102 assert "q" in schema["function"]["parameters"]["properties"] 

103 

104 def test_to_anthropic_schema(self): 

105 client = MCPClient() 

106 tool = MCPToolInfo(name="run", description="Run command", server_name="shell") 

107 adapter = MCPToolAdapter(client=client, tool_info=tool) 

108 schema = adapter.to_anthropic_schema() 

109 assert schema["name"] == "mcp__shell__run" 

110 

111 def test_write_operation_detection(self): 

112 client = MCPClient() 

113 tool = MCPToolInfo(name="write_file", server_name="fs") 

114 adapter = MCPToolAdapter(client=client, tool_info=tool) 

115 assert adapter.is_write_operation({"path": "/tmp/x"}) 

116 assert not adapter.is_read_operation({"path": "/tmp/x"}) 

117 

118 def test_read_operation_detection(self): 

119 client = MCPClient() 

120 tool = MCPToolInfo(name="read_file", server_name="fs") 

121 adapter = MCPToolAdapter(client=client, tool_info=tool) 

122 assert not adapter.is_write_operation({"path": "/tmp/x"}) 

123 assert adapter.is_read_operation({"path": "/tmp/x"}) 

124 

125 def test_extract_target_path(self): 

126 client = MCPClient() 

127 tool = MCPToolInfo(name="tool", server_name="s") 

128 adapter = MCPToolAdapter(client=client, tool_info=tool) 

129 assert adapter.extract_target_path({"path": "/a/b"}) == "/a/b" 

130 assert adapter.extract_target_path({"uri": "file:///x"}) == "file:///x" 

131 

132 def test_permission_default(self): 

133 client = MCPClient() 

134 tool = MCPToolInfo(name="t", server_name="s") 

135 adapter = MCPToolAdapter(client=client, tool_info=tool) 

136 assert adapter.permission_level == PermissionLevel.MODERATE 

137 

138 def test_permission_custom(self): 

139 client = MCPClient() 

140 tool = MCPToolInfo(name="t", server_name="s") 

141 adapter = MCPToolAdapter( 

142 client=client, 

143 tool_info=tool, 

144 permission_level=PermissionLevel.SAFE, 

145 ) 

146 assert adapter.permission_level == PermissionLevel.SAFE 

147 

148 

149class TestMCPToolRegistry: 

150 """Tool registry tests.""" 

151 

152 def test_empty_registry(self): 

153 client = MCPClient() 

154 registry = MCPToolRegistry(client) 

155 assert registry.get_all_tools() == {} 

156 assert registry.get_tool("nonexistent") is None 

157 

158 def test_refresh(self): 

159 client = MCPClient() 

160 registry = MCPToolRegistry(client) 

161 registry.refresh() # Should not raise 

162 

163 

164class TestMCPDataModels: 

165 """Data model tests.""" 

166 

167 def test_tool_info_minimal(self): 

168 info = MCPToolInfo(name="t", server_name="s") 

169 assert info.description == "" 

170 assert info.input_schema == {} 

171 

172 def test_resource_info(self): 

173 info = MCPResourceInfo( 

174 uri="file:///data", 

175 name="config", 

176 mime_type="application/json", 

177 server_name="s", 

178 ) 

179 assert info.uri == "file:///data" 

180 assert info.mime_type == "application/json" 

181 

182 def test_prompt_info(self): 

183 info = MCPPromptInfo( 

184 name="greet", 

185 description="Generate greeting", 

186 arguments=[{"name": "style", "required": True}], 

187 server_name="s", 

188 ) 

189 assert len(info.arguments) == 1 

190 assert info.arguments[0]["required"] 

191 

192 

193class TestMCPError: 

194 """Error handling tests.""" 

195 

196 def test_error_basic(self): 

197 err = MCPError(-32602, "Invalid params") 

198 assert err.code == -32602 

199 assert "Invalid params" in str(err) 

200 

201 def test_error_with_data(self): 

202 err = MCPError(-1, "custom", data={"detail": "xyz"}) 

203 assert err.data == {"detail": "xyz"} 

204 

205 

206class TestMCPToolAdapterEdgeCases: 

207 """Edge case tests for adapter behavior.""" 

208 

209 def test_adapter_empty_schema(self): 

210 client = MCPClient() 

211 tool = MCPToolInfo(name="empty", server_name="s") 

212 adapter = MCPToolAdapter(client=client, tool_info=tool) 

213 schema = adapter.to_openai_schema() 

214 assert "properties" in schema["function"]["parameters"] 

215 

216 def test_adapter_no_description(self): 

217 client = MCPClient() 

218 tool = MCPToolInfo(name="t", server_name="s") 

219 adapter = MCPToolAdapter(client=client, tool_info=tool) 

220 assert "mcp" in adapter.description.lower()