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

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

2 

3import pytest 

4from agentos.mcp import ( 

5 MCPClient, 

6 MCPServerConfig, 

7 MCPToolInfo, 

8 MCPResourceInfo, 

9 MCPPromptInfo, 

10 MCPError, 

11) 

12from agentos.mcp.adapter import MCPToolAdapter, MCPToolRegistry 

13from agentos.tools.base import PermissionLevel, ToolResult 

14 

15 

16class TestMCPServerConfig: 

17 """Server configuration tests.""" 

18 

19 def test_defaults(self): 

20 config = MCPServerConfig(name="test") 

21 assert config.name == "test" 

22 assert config.transport == "stdio" 

23 assert config.args == [] 

24 assert config.timeout == 30 

25 

26 def test_custom(self): 

27 config = MCPServerConfig( 

28 name="github", 

29 transport="sse", 

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

31 timeout=60, 

32 ) 

33 assert config.transport == "sse" 

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

35 assert config.timeout == 60 

36 

37 

38class TestMCPClientLifecycle: 

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

40 

41 @pytest.mark.asyncio 

42 async def test_init_empty(self): 

43 client = MCPClient() 

44 assert client.connected_servers == [] 

45 assert client.list_tools() == [] 

46 assert client.list_resources() == [] 

47 assert client.list_prompts() == [] 

48 

49 @pytest.mark.asyncio 

50 async def test_context_manager(self): 

51 async with MCPClient() as client: 

52 assert client.connected_servers == [] 

53 

54 @pytest.mark.asyncio 

55 async def test_connect_unknown_transport(self): 

56 client = MCPClient() 

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

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

59 await client.connect_server(config) 

60 

61 @pytest.mark.asyncio 

62 async def test_sse_requires_url(self): 

63 client = MCPClient() 

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

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

66 await client.connect_server(config) 

67 

68 

69class TestMCPToolAdapter: 

70 """Tool adapter wrapping tests.""" 

71 

72 def test_adapt_tool_basic(self): 

73 client = MCPClient() 

74 tool = MCPToolInfo( 

75 name="read", 

76 description="Read a file", 

77 server_name="fs", 

78 input_schema={ 

79 "type": "object", 

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

81 "required": ["path"], 

82 }, 

83 ) 

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

85 assert adapter.name == "mcp__fs__read" 

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

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

88 

89 def test_to_openai_schema(self): 

90 client = MCPClient() 

91 tool = MCPToolInfo( 

92 name="search", 

93 description="Search docs", 

94 server_name="docs", 

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

96 ) 

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

98 schema = adapter.to_openai_schema() 

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

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

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

102 

103 def test_to_anthropic_schema(self): 

104 client = MCPClient() 

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

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

107 schema = adapter.to_anthropic_schema() 

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

109 

110 def test_write_operation_detection(self): 

111 client = MCPClient() 

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

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

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

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

116 

117 def test_read_operation_detection(self): 

118 client = MCPClient() 

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

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

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

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

123 

124 def test_extract_target_path(self): 

125 client = MCPClient() 

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

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

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

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

130 

131 def test_permission_default(self): 

132 client = MCPClient() 

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

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

135 assert adapter.permission_level == PermissionLevel.MODERATE 

136 

137 def test_permission_custom(self): 

138 client = MCPClient() 

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

140 adapter = MCPToolAdapter( 

141 client=client, 

142 tool_info=tool, 

143 permission_level=PermissionLevel.SAFE, 

144 ) 

145 assert adapter.permission_level == PermissionLevel.SAFE 

146 

147 

148class TestMCPToolRegistry: 

149 """Tool registry tests.""" 

150 

151 def test_empty_registry(self): 

152 client = MCPClient() 

153 registry = MCPToolRegistry(client) 

154 assert registry.get_all_tools() == {} 

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

156 

157 def test_refresh(self): 

158 client = MCPClient() 

159 registry = MCPToolRegistry(client) 

160 registry.refresh() # Should not raise 

161 

162 

163class TestMCPDataModels: 

164 """Data model tests.""" 

165 

166 def test_tool_info_minimal(self): 

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

168 assert info.description == "" 

169 assert info.input_schema == {} 

170 

171 def test_resource_info(self): 

172 info = MCPResourceInfo( 

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

174 name="config", 

175 mime_type="application/json", 

176 server_name="s", 

177 ) 

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

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

180 

181 def test_prompt_info(self): 

182 info = MCPPromptInfo( 

183 name="greet", 

184 description="Generate greeting", 

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

186 server_name="s", 

187 ) 

188 assert len(info.arguments) == 1 

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

190 

191 

192class TestMCPError: 

193 """Error handling tests.""" 

194 

195 def test_error_basic(self): 

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

197 assert err.code == -32602 

198 assert "Invalid params" in str(err) 

199 

200 def test_error_with_data(self): 

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

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

203 

204 

205class TestMCPToolAdapterEdgeCases: 

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

207 

208 def test_adapter_empty_schema(self): 

209 client = MCPClient() 

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

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

212 schema = adapter.to_openai_schema() 

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

214 

215 def test_adapter_no_description(self): 

216 client = MCPClient() 

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

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

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