Update mcp.py · XFE Studio Git

@@ -7,9 +7,9 @@ from pathlib import Path

7 7

8 8 from g4f.mcp.server import MCPServer, MCPRequest

9 9 from g4f.mcp.tools import (

10 WebSearchTool, WebScrapeTool, ImageGenerationTool,

11 PythonExecuteTool, FileReadTool, FileReadLinesTool, FileSearchTool,

12 FileWriteTool, FileListTool, FileDeleteTool,

10 WebSearchTool, ImageGenerationTool,

11 PythonExecuteTool, FileReadTool,

12 FileListTool, FileDeleteTool,

13 13 )

14 14 from g4f.mcp.pa_provider import execute_safe_code, get_workspace_dir, SAFE_MODULES

15 15

@@ -34,13 +34,9 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

34 34 self.assertEqual(server.server_info["name"], "gpt4free-mcp-server")

35 35 self.assertEqual(len(server.tools), _TOOL_COUNT)

36 36 self.assertIn('web_search', server.tools)

37 self.assertIn('web_scrape', server.tools)

38 37 self.assertIn('image_generation', server.tools)

39 38 self.assertIn('python_execute', server.tools)

40 39 self.assertIn('file_read', server.tools)

41 self.assertIn('file_read_lines', server.tools)

42 self.assertIn('file_search', server.tools)

43 self.assertIn('file_write', server.tools)

44 40 self.assertIn('file_list', server.tools)

45 41 self.assertIn('file_delete', server.tools)

46 42

@@ -83,13 +79,9 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

83 79

84 80 tool_names = [tool["name"] for tool in response.result["tools"]]

85 81 self.assertIn("web_search", tool_names)

86 self.assertIn("web_scrape", tool_names)

87 82 self.assertIn("image_generation", tool_names)

88 83 self.assertIn("python_execute", tool_names)

89 84 self.assertIn("file_read", tool_names)

90 self.assertIn("file_read_lines", tool_names)

91 self.assertIn("file_search", tool_names)

92 self.assertIn("file_write", tool_names)

93 85 self.assertIn("file_list", tool_names)

94 86 self.assertIn("file_delete", tool_names)

95 87

@@ -156,13 +148,6 @@ class TestMCPTools(unittest.IsolatedAsyncioTestCase):

156 148 self.assertIn("query", tool.input_schema["properties"])

157 149 self.assertIn("query", tool.input_schema["required"])

158 150

159 async def test_web_scrape_tool_schema(self):

160 tool = WebScrapeTool()

161 self.assertIsNotNone(tool.description)

162 self.assertIsNotNone(tool.input_schema)

163 self.assertEqual(tool.input_schema["type"], "object")

164 self.assertIn("url", tool.input_schema["properties"])

165 self.assertIn("url", tool.input_schema["required"])

166 151

167 152 async def test_image_generation_tool_schema(self):

168 153 tool = ImageGenerationTool()

@@ -177,11 +162,6 @@ class TestMCPTools(unittest.IsolatedAsyncioTestCase):

177 162 result = await tool.execute({})

178 163 self.assertIn("error", result)

179 164

180 async def test_web_scrape_missing_url(self):

181 tool = WebScrapeTool()

182 result = await tool.execute({})

183 self.assertIn("error", result)

184

185 165 async def test_image_generation_missing_prompt(self):

186 166 tool = ImageGenerationTool()

187 167 result = await tool.execute({})

@@ -272,11 +252,6 @@ class TestFilesTools(unittest.IsolatedAsyncioTestCase):

272 252 schema = tool.input_schema

273 253 self.assertEqual(schema["type"], "object")

274 254

275 async def test_file_write_schema(self):

276 tool = FileWriteTool()

277 schema = tool.input_schema

278 self.assertIn("path", schema["properties"])

279 self.assertIn("content", schema["properties"])

280 255

281 256 async def test_file_read_schema(self):

282 257 tool = FileReadTool()

@@ -290,21 +265,6 @@ class TestFilesTools(unittest.IsolatedAsyncioTestCase):

290 265 self.assertIn("path", schema["properties"])

291 266 self.assertIn("path", schema["required"])

292 267

293 async def test_write_and_read(self):

294 write_tool = FileWriteTool()

295 read_tool = FileReadTool()

296

297 write_result = await write_tool.execute({

298 "path": self.test_file,

299 "content": "hello workspace",

300 })

301 self.assertNotIn("error", write_result)

302 self.assertEqual(write_result["path"], self.test_file)

303

304 read_result = await read_tool.execute({"path": self.test_file})

305 self.assertNotIn("error", read_result)

306 self.assertEqual(read_result["content"], "hello workspace")

307

308 268 async def test_read_missing_file(self):

309 269 tool = FileReadTool()

310 270 result = await tool.execute({"path": "definitely_does_not_exist.txt"})

@@ -315,17 +275,6 @@ class TestFilesTools(unittest.IsolatedAsyncioTestCase):

315 275 result = await tool.execute({})

316 276 self.assertIn("error", result)

317 277

318 async def test_write_and_delete(self):

319 write_tool = FileWriteTool()

320 delete_tool = FileDeleteTool()

321

322 await write_tool.execute({"path": self.test_file, "content": "to delete"})

323 self.assertTrue((self.workspace / self.test_file).exists())

324

325 delete_result = await delete_tool.execute({"path": self.test_file})

326 self.assertNotIn("error", delete_result)

327 self.assertTrue(delete_result.get("deleted"))

328 self.assertFalse((self.workspace / self.test_file).exists())

329 278

330 279 async def test_delete_missing_file(self):

331 280 tool = FileDeleteTool()

@@ -350,27 +299,6 @@ class TestFilesTools(unittest.IsolatedAsyncioTestCase):

350 299 result = await read_tool.execute({"path": "../../etc/passwd"})

351 300 self.assertIn("error", result)

352 301

353 async def test_file_read_lines(self):

354 write_tool = FileWriteTool()

355 read_lines_tool = FileReadLinesTool()

356

357 await write_tool.execute({"path": self.test_file, "content": "line1\nline2\nline3\n"})

358 result = await read_lines_tool.execute({"path": self.test_file, "start_line": 2, "end_line": 3})

359

360 self.assertNotIn("error", result)

361 self.assertEqual(result["returned_lines"], 2)

362 self.assertEqual(result["lines"], ["line2\n", "line3\n"])

363

364 async def test_file_search_by_pattern_and_content(self):

365 write_tool = FileWriteTool()

366 search_tool = FileSearchTool()

367

368 await write_tool.execute({"path": self.test_file, "content": "hello workspace"})

369 result = await search_tool.execute({"pattern": "*test.txt", "query": "workspace"})

370

371 self.assertNotIn("error", result)

372 self.assertTrue(result["count"] >= 1)

373 self.assertTrue(any(self.test_file in match["path"] for match in result["matches"]))

374 302

375 303 async def test_file_append(self):

376 304 write_tool = FileWriteTool()

代码差异