gpt4free

公开

关注 0 Fork 0 Star 1

XFEstudio/gpt4free

Add .pa.py safe executor, workspace file tools, and MCP server integration

Agent-Logs-Url: https://github.com/xtekky/gpt4free/sessions/1b3f481e-143f-4c30-9982-1063c0338ec3 Co-authored-by: hlohaus <983577+hlohaus@users.noreply.github.com>

a5b4675d

copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>

提交于 2026-04-04 16:19

父提交 9d4cbff

代码差异

5 个文件 +941 -33

Modified etc/unittest/mcp.py +266 -25

@@ -1,10 +1,16 @@

1 1 from __future__ import annotations

2 2

3 3 import json

4 import os

4 5 import unittest

6 from pathlib import Path

5 7

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

7 from g4f.mcp.tools import WebSearchTool, WebScrapeTool, ImageGenerationTool

9 from g4f.mcp.tools import (

10 WebSearchTool, WebScrapeTool, ImageGenerationTool,

11 PythonExecuteTool, FileReadTool, FileWriteTool, FileListTool, FileDeleteTool,

12 )

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

8 14

9 15 try:

10 16 from ddgs import DDGS, DDGSError

@@ -13,20 +19,28 @@ try:

13 19 except ImportError:

14 20 has_requirements = False

15 21

22 # Total number of tools registered in MCPServer

23 _TOOL_COUNT = 10

16 25

17 26 class TestMCPServer(unittest.IsolatedAsyncioTestCase):

18 27 """Test cases for MCP server"""

20 29 async def test_server_initialization(self):

21 30 """Test that server initializes correctly"""

22 31 server = MCPServer()

23 32 self.assertIsNotNone(server)

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

25 self.assertEqual(len(server.tools), 5)

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

26 35 self.assertIn('web_search', server.tools)

27 36 self.assertIn('web_scrape', server.tools)

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

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

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

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

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

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

30 44 async def test_initialize_request(self):

31 45 """Test initialize method"""

32 46 server = MCPServer()

@@ -42,7 +56,7 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

42 56 self.assertIsNotNone(response.result)

43 57 self.assertEqual(response.result["protocolVersion"], "2024-11-05")

44 58 self.assertIn("serverInfo", response.result)

46 60 async def test_tools_list(self):

47 61 """Test tools/list method"""

48 62 server = MCPServer()

@@ -57,14 +71,18 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

57 71 self.assertEqual(response.id, 2)

58 72 self.assertIsNotNone(response.result)

59 73 self.assertIn("tools", response.result)

60 self.assertEqual(len(response.result["tools"]), 5)

62 # Check tool structure

74 self.assertEqual(len(response.result["tools"]), _TOOL_COUNT)

63 76 tool_names = [tool["name"] for tool in response.result["tools"]]

64 77 self.assertIn("web_search", tool_names)

65 78 self.assertIn("web_scrape", tool_names)

66 79 self.assertIn("image_generation", tool_names)

80 self.assertIn("python_execute", tool_names)

81 self.assertIn("file_read", tool_names)

82 self.assertIn("file_write", tool_names)

83 self.assertIn("file_list", tool_names)

84 self.assertIn("file_delete", tool_names)

68 86 async def test_ping(self):

69 87 """Test ping method"""

70 88 server = MCPServer()

@@ -78,7 +96,7 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

78 96 self.assertEqual(response.jsonrpc, "2.0")

79 97 self.assertEqual(response.id, 3)

80 98 self.assertIsNotNone(response.result)

82 100 async def test_invalid_method(self):

83 101 """Test invalid method returns error"""

84 102 server = MCPServer()

@@ -93,7 +111,7 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

93 111 self.assertEqual(response.id, 4)

94 112 self.assertIsNotNone(response.error)

95 113 self.assertEqual(response.error["code"], -32601)

114

97 115 async def test_tool_call_invalid_tool(self):

98 116 """Test calling non-existent tool"""

99 117 server = MCPServer()

@@ -114,53 +132,276 @@ class TestMCPServer(unittest.IsolatedAsyncioTestCase):

114 132

115 133

116 134 class TestMCPTools(unittest.IsolatedAsyncioTestCase):

117 """Test cases for MCP tools"""

118

135 """Test cases for existing MCP tools"""

136

119 137 def setUp(self) -> None:

120 138 if not has_requirements:

121 139 self.skipTest('MCP tools requirements not installed')

122

140

123 141 async def test_web_search_tool_schema(self):

124 """Test WebSearchTool schema"""

125 142 tool = WebSearchTool()

126 143 self.assertIsNotNone(tool.description)

127 144 self.assertIsNotNone(tool.input_schema)

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

129 146 self.assertIn("query", tool.input_schema["properties"])

130 147 self.assertIn("query", tool.input_schema["required"])

131

148

132 149 async def test_web_scrape_tool_schema(self):

133 """Test WebScrapeTool schema"""

134 150 tool = WebScrapeTool()

135 151 self.assertIsNotNone(tool.description)

136 152 self.assertIsNotNone(tool.input_schema)

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

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

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

140

156

141 157 async def test_image_generation_tool_schema(self):

142 """Test ImageGenerationTool schema"""

143 158 tool = ImageGenerationTool()

144 159 self.assertIsNotNone(tool.description)

145 160 self.assertIsNotNone(tool.input_schema)

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

147 162 self.assertIn("prompt", tool.input_schema["properties"])

148 163 self.assertIn("prompt", tool.input_schema["required"])

149

164

150 165 async def test_web_search_missing_query(self):

151 """Test web search with missing query parameter"""

152 166 tool = WebSearchTool()

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

154 168 self.assertIn("error", result)

155

169

156 170 async def test_web_scrape_missing_url(self):

157 """Test web scrape with missing url parameter"""

158 171 tool = WebScrapeTool()

159 172 result = await tool.execute({})

160 173 self.assertIn("error", result)

161

174

162 175 async def test_image_generation_missing_prompt(self):

163 """Test image generation with missing prompt parameter"""

164 176 tool = ImageGenerationTool()

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

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

179

180

181 class TestPythonExecuteTool(unittest.IsolatedAsyncioTestCase):

182 """Tests for the PythonExecuteTool (no network required)."""

183

184 async def test_schema(self):

185 tool = PythonExecuteTool()

186 self.assertIsNotNone(tool.description)

187 schema = tool.input_schema

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

189 self.assertIn("code", schema["properties"])

190 self.assertIn("code", schema["required"])

191

192 async def test_missing_code(self):

193 tool = PythonExecuteTool()

194 result = await tool.execute({})

195 self.assertIn("error", result)

196

197 async def test_simple_execution(self):

198 tool = PythonExecuteTool()

199 result = await tool.execute({"code": "result = 1 + 2"})

200 self.assertTrue(result.get("success"))

201 self.assertEqual(result.get("result"), 3)

202

203 async def test_stdout_captured(self):

204 tool = PythonExecuteTool()

205 result = await tool.execute({"code": "print('hello')"})

206 self.assertTrue(result.get("success"))

207 self.assertIn("hello", result.get("stdout", ""))

208

209 async def test_syntax_error(self):

210 tool = PythonExecuteTool()

211 result = await tool.execute({"code": "def foo(:"})

212 self.assertFalse(result.get("success"))

213 self.assertIn("error", result)

214

215 async def test_blocked_import(self):

216 tool = PythonExecuteTool()

217 result = await tool.execute({"code": "import subprocess"})

218 self.assertFalse(result.get("success"))

219 self.assertIn("error", result)

220

221 async def test_blocked_builtin_exec(self):

222 """exec() is removed from safe builtins."""

223 tool = PythonExecuteTool()

224 result = await tool.execute({"code": "exec('x=1')"})

225 self.assertFalse(result.get("success"))

226 self.assertIn("error", result)

227

228 async def test_allowed_module(self):

229 tool = PythonExecuteTool()

230 result = await tool.execute({"code": "import math\nresult = math.sqrt(4)"})

231 self.assertTrue(result.get("success"))

232 self.assertAlmostEqual(result.get("result"), 2.0)

233

234 async def test_allowed_json_module(self):

235 tool = PythonExecuteTool()

236 result = await tool.execute({

237 "code": "import json\nresult = json.dumps({'a': 1})"

238 })

239 self.assertTrue(result.get("success"))

240 self.assertEqual(result.get("result"), '{"a": 1}')

241

242

243 class TestFilesTools(unittest.IsolatedAsyncioTestCase):

244 """Tests for the file manipulation tools (workspace)."""

245

246 def setUp(self):

247 self.workspace = get_workspace_dir()

248 self.test_file = "unittest_temp_test.txt"

249 # Clean up leftover test file

250 target = self.workspace / self.test_file

251 if target.exists():

252 target.unlink()

253

254 def tearDown(self):

255 target = self.workspace / self.test_file

256 if target.exists():

257 target.unlink()

258

259 async def test_file_list_schema(self):

260 tool = FileListTool()

261 self.assertIsNotNone(tool.description)

262 schema = tool.input_schema

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

264

265 async def test_file_write_schema(self):

266 tool = FileWriteTool()

267 schema = tool.input_schema

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

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

270

271 async def test_file_read_schema(self):

272 tool = FileReadTool()

273 schema = tool.input_schema

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

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

276

277 async def test_file_delete_schema(self):

278 tool = FileDeleteTool()

279 schema = tool.input_schema

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

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

282

283 async def test_write_and_read(self):

284 write_tool = FileWriteTool()

285 read_tool = FileReadTool()

286

287 write_result = await write_tool.execute({

288 "path": self.test_file,

289 "content": "hello workspace",

290 })

291 self.assertNotIn("error", write_result)

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

293

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

295 self.assertNotIn("error", read_result)

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

297

298 async def test_read_missing_file(self):

299 tool = FileReadTool()

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

301 self.assertIn("error", result)

302

303 async def test_read_missing_path_param(self):

304 tool = FileReadTool()

305 result = await tool.execute({})

306 self.assertIn("error", result)

307

308 async def test_write_and_delete(self):

309 write_tool = FileWriteTool()

310 delete_tool = FileDeleteTool()

311

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

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

314

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

316 self.assertNotIn("error", delete_result)

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

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

319

320 async def test_delete_missing_file(self):

321 tool = FileDeleteTool()

322 result = await tool.execute({"path": "no_such_file.txt"})

323 self.assertIn("error", result)

324

325 async def test_list_workspace(self):

326 # Write a temp file so workspace is non-empty

327 write_tool = FileWriteTool()

328 await write_tool.execute({"path": self.test_file, "content": "x"})

329

330 list_tool = FileListTool()

331 result = await list_tool.execute({})

332 self.assertNotIn("error", result)

333 self.assertIn("entries", result)

334 paths = [e["path"] for e in result["entries"]]

335 self.assertIn(self.test_file, paths)

336

337 async def test_path_traversal_blocked(self):

338 """Ensure path traversal outside workspace is rejected."""

339 read_tool = FileReadTool()

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

341 self.assertIn("error", result)

342

343 async def test_file_append(self):

344 write_tool = FileWriteTool()

345 read_tool = FileReadTool()

346

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

348 await write_tool.execute({"path": self.test_file, "content": "line2\n", "append": True})

349

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

351 self.assertEqual(read_result["content"], "line1\nline2\n")

352

353

354 class TestSafeCodeExecution(unittest.TestCase):

355 """Unit tests for the execute_safe_code() function directly."""

356

357 def test_basic_result(self):

358 r = execute_safe_code("result = 42")

359 self.assertTrue(r.success)

360 self.assertEqual(r.result, 42)

361

362 def test_stdout(self):

363 r = execute_safe_code("print('hi')")

364 self.assertTrue(r.success)

365 self.assertIn("hi", r.stdout)

366

367 def test_runtime_error(self):

368 r = execute_safe_code("1/0")

369 self.assertFalse(r.success)

370 self.assertIn("ZeroDivisionError", r.error)

371

372 def test_blocked_os_import(self):

373 r = execute_safe_code("import os")

374 self.assertFalse(r.success)

375

376 def test_blocked_sys_import(self):

377 r = execute_safe_code("import sys")

378 self.assertFalse(r.success)

379

380 def test_blocked_subprocess(self):

381 r = execute_safe_code("import subprocess")

382 self.assertFalse(r.success)

383

384 def test_allowed_math(self):

385 r = execute_safe_code("import math\nresult = math.pi")

386 self.assertTrue(r.success)

387 self.assertAlmostEqual(r.result, 3.14159, places=4)

388

389 def test_to_dict_success(self):

390 r = execute_safe_code("result = [1, 2, 3]")

391 d = r.to_dict()

392 self.assertTrue(d["success"])

393 self.assertEqual(d["result"], [1, 2, 3])

394

395 def test_to_dict_failure(self):

396 r = execute_safe_code("raise ValueError('boom')")

397 d = r.to_dict()

398 self.assertFalse(d["success"])

399 self.assertIn("error", d)

400

401 def test_safe_modules_frozenset(self):

402 self.assertIsInstance(SAFE_MODULES, frozenset)

403 self.assertIn("math", SAFE_MODULES)

404 self.assertIn("json", SAFE_MODULES)

405 self.assertIn("asyncio", SAFE_MODULES)

406 self.assertNotIn("os", SAFE_MODULES)

407 self.assertNotIn("subprocess", SAFE_MODULES)

Modified g4f/mcp/__init__.py +46 -3

@@ -3,11 +3,54 @@

3 3 This module provides an MCP server implementation that exposes gpt4free capabilities

4 4 through the Model Context Protocol standard, allowing AI assistants to access:

5 5 - Web search functionality

6 - Web scraping capabilities

7 7 - Image generation using various providers

8 - Safe Python code execution

9 - Workspace file management (read, write, list, delete) at ~/.g4f/workspace

10 - .pa.py custom provider loading and execution

8 11 """

9 12

10 13 from .server import MCPServer

11 from .tools import MarkItDownTool, TextToAudioTool, WebSearchTool, WebScrapeTool, ImageGenerationTool

14 from .tools import (

15 MarkItDownTool,

16 TextToAudioTool,

17 WebSearchTool,

18 WebScrapeTool,

19 ImageGenerationTool,

20 PythonExecuteTool,

21 FileReadTool,

22 FileWriteTool,

23 FileListTool,

24 FileDeleteTool,

25 )

26 from .pa_provider import (

27 execute_safe_code,

28 load_pa_provider,

29 list_pa_providers,

30 get_workspace_dir,

31 SAFE_MODULES,

32 SafeExecutionResult,

33 )

12 34

13 __all__ = ['MCPServer', 'MarkItDownTool', 'TextToAudioTool', 'WebSearchTool', 'WebScrapeTool', 'ImageGenerationTool']

35 __all__ = [

36 'MCPServer',

37 # Original tools

38 'MarkItDownTool',

39 'TextToAudioTool',

40 'WebSearchTool',

41 'WebScrapeTool',

42 'ImageGenerationTool',

43 # New tools

44 'PythonExecuteTool',

45 'FileReadTool',

46 'FileWriteTool',

47 'FileListTool',

48 'FileDeleteTool',

49 # PA provider system

50 'execute_safe_code',

51 'load_pa_provider',

52 'list_pa_providers',

53 'get_workspace_dir',

54 'SAFE_MODULES',

55 'SafeExecutionResult',

56 ]

Added g4f/mcp/pa_provider.py +316 -0

@@ -0,0 +1,316 @@

1 """PA Provider - .pa.py file parser and executor for custom providers

3 This module provides:

5 1. Safe Python code execution with whitelisted modules and restricted built-ins.

6 2. ``.pa.py`` file loading — parse and execute provider-adapter files that define

7 custom gpt4free providers.

8 3. Workspace management at ``~/.g4f/workspace``.

10 A ``.pa.py`` file is a plain Python file that is executed inside a sandbox.

11 Inside that sandbox the code may only import from the *whitelisted* module set

12 and may only access the file-system through a workspace-scoped ``open()``.

14 Typical layout of a ``.pa.py`` file::

16 from aiohttp import ClientSession

17 from g4f.Provider.base_provider import AsyncGeneratorProvider, ProviderModelMixin

18 from g4f.Provider.helper import format_prompt

19 from g4f.typing import AsyncResult, Messages

21 class Provider(AsyncGeneratorProvider, ProviderModelMixin):

22 label = "MyCustomProvider"

23 url = "https://example.com"

24 working = True

25 default_model = "gpt-4"

26 models = ["gpt-4", "gpt-3.5-turbo"]

28 @classmethod

29 async def create_async_generator(cls, model, messages, **kwargs):

30 ...

31 yield chunk

32 """

34 from __future__ import annotations

36 import io

37 import ast

38 import json

39 import contextlib

40 import traceback

41 import builtins as _builtins

42 from pathlib import Path

43 from typing import Any, Dict, FrozenSet, List, Optional

45 # ---------------------------------------------------------------------------

46 # Workspace directory

47 # ---------------------------------------------------------------------------

49 def get_workspace_dir() -> Path:

50 """Return the workspace directory ``~/.g4f/workspace``, creating it if needed."""

51 workspace = Path.home() / ".g4f" / "workspace"

52 workspace.mkdir(parents=True, exist_ok=True)

53 return workspace

56 # ---------------------------------------------------------------------------

57 # Whitelisted modules

58 # ---------------------------------------------------------------------------

60 #: Modules that are allowed inside the safe execution sandbox.

61 SAFE_MODULES: FrozenSet[str] = frozenset({

62 # Math / numeric

63 "math", "cmath", "decimal", "fractions", "statistics", "random", "numbers",

64 # String / text

65 "string", "re", "textwrap", "unicodedata", "difflib", "fnmatch",

66 # Data structures

67 "json", "csv", "collections", "heapq", "bisect", "array", "queue",

68 # Functional

69 "itertools", "functools", "operator",

70 # Type system

71 "typing", "types", "abc", "dataclasses", "enum",

72 # Time / date

73 "datetime", "time", "calendar",

74 # I/O

75 "io", "pathlib",

76 # Async

77 "asyncio",

78 # Encoding / hashing

79 "base64", "hashlib", "hmac", "binascii", "codecs", "struct",

80 # URL / HTTP

81 "urllib", "urllib.parse", "http", "http.client",

82 # Compression

83 "gzip", "zlib",

84 # Misc safe stdlib

85 "copy", "pprint", "reprlib", "warnings", "contextlib",

86 # Third-party HTTP (used by providers)

87 "aiohttp", "requests",

88 # gpt4free itself

89 "g4f",

90 })

93 # ---------------------------------------------------------------------------

94 # Sandbox helpers

95 # ---------------------------------------------------------------------------

97 def _make_restricted_import(allowed: FrozenSet[str]):

98 """Return a ``__import__`` replacement that only allows *allowed* modules."""

99 original = _builtins.__import__

100

101 def _restricted_import(name, globals=None, locals=None, fromlist=(), level=0):

102 if level > 0:

103 raise ImportError(

104 "Relative imports are not allowed inside a .pa.py sandbox."

105 )

106 base = name.split(".")[0]

107 if base not in allowed:

108 raise ImportError(

109 f"Import of '{name}' is not allowed in safe execution mode.\n"

110 f"Allowed top-level modules: {', '.join(sorted(allowed))}"

111 )

112 return original(name, globals, locals, fromlist, level)

113

114 return _restricted_import

115

116

117 def _make_safe_globals(allowed: FrozenSet[str] = SAFE_MODULES) -> Dict[str, Any]:

118 """Return a ``globals`` dict suitable for sandboxed ``exec``."""

119 workspace = get_workspace_dir()

120

121 # Build a reduced copy of the real built-ins

122 _blocked = frozenset({"exec", "eval", "compile", "input", "breakpoint", "__import__"})

123 safe_builtins: Dict[str, Any] = {

124 k: getattr(_builtins, k)

125 for k in dir(_builtins)

126 if k not in _blocked

127 }

128

129 # Provide a workspace-scoped open()

130 def _safe_open(file, mode="r", *args, **kwargs):

131 """open() restricted to the workspace directory."""

132 path = Path(file)

133 if not path.is_absolute():

134 path = workspace / path

135 try:

136 resolved = path.resolve()

137 ws_resolved = workspace.resolve()

138 if not str(resolved).startswith(str(ws_resolved)):

139 raise PermissionError(

140 f"File access outside workspace is denied: '{file}'. "

141 f"Workspace: {workspace}"

142 )

143 except (ValueError, OSError) as exc:

144 raise PermissionError(f"Invalid file path: '{file}'") from exc

145 return open(resolved, mode, *args, **kwargs)

146

147 safe_builtins["open"] = _safe_open

148 safe_builtins["__import__"] = _make_restricted_import(allowed)

149

150 return {

151 "__builtins__": safe_builtins,

152 "__name__": "__pa_provider__",

153 }

154

155

156 # ---------------------------------------------------------------------------

157 # Execution result

158 # ---------------------------------------------------------------------------

159

160 class SafeExecutionResult:

161 """Holds the outcome of a sandboxed code execution."""

162

163 def __init__(

164 self,

165 success: bool,

166 stdout: str = "",

167 stderr: str = "",

168 result: Any = None,

169 error: Optional[str] = None,

170 locals: Optional[Dict[str, Any]] = None,

171 ) -> None:

172 self.success = success

173 self.stdout = stdout

174 self.stderr = stderr

175 self.result = result

176 self.error = error

177 self.locals: Dict[str, Any] = locals or {}

178

179 def to_dict(self) -> Dict[str, Any]:

180 data: Dict[str, Any] = {

181 "success": self.success,

182 "stdout": self.stdout,

183 "stderr": self.stderr,

184 }

185 if self.error:

186 data["error"] = self.error

187 if self.result is not None:

188 try:

189 json.dumps(self.result)

190 data["result"] = self.result

191 except (TypeError, ValueError):

192 data["result"] = repr(self.result)

193 return data

194

195

196 # ---------------------------------------------------------------------------

197 # Safe executor

198 # ---------------------------------------------------------------------------

199

200 def execute_safe_code(

201 code: str,

202 extra_globals: Optional[Dict[str, Any]] = None,

203 allowed_modules: FrozenSet[str] = SAFE_MODULES,

204 ) -> SafeExecutionResult:

205 """Execute *code* inside a safe sandbox with whitelisted module imports.

206

207 Args:

208 code: Python source code to execute.

209 extra_globals: Additional names injected into the execution globals.

210 allowed_modules: Frozenset of top-level module names that may be imported.

211

212 Returns:

213 :class:`SafeExecutionResult` containing captured stdout/stderr, any

214 ``result`` variable assigned in the code, or error information.

215 """

216 stdout_buf = io.StringIO()

217 stderr_buf = io.StringIO()

218

219 safe_globals = _make_safe_globals(allowed_modules)

220 if extra_globals:

221 safe_globals.update(extra_globals)

222

223 local_vars: Dict[str, Any] = {}

224

225 try:

226 compiled = compile(code, "<pa_provider>", "exec")

227 with (

228 contextlib.redirect_stdout(stdout_buf),

229 contextlib.redirect_stderr(stderr_buf),

230 ):

231 exec(compiled, safe_globals, local_vars) # noqa: S102

232

233 return SafeExecutionResult(

234 success=True,

235 stdout=stdout_buf.getvalue(),

236 stderr=stderr_buf.getvalue(),

237 result=local_vars.get("result"),

238 locals=local_vars,

239 )

240

241 except Exception:

242 return SafeExecutionResult(

243 success=False,

244 stdout=stdout_buf.getvalue(),

245 stderr=stderr_buf.getvalue(),

246 error=traceback.format_exc(),

247 )

248

249

250 # ---------------------------------------------------------------------------

251 # .pa.py provider loader

252 # ---------------------------------------------------------------------------

253

254 def load_pa_provider(file_path: "str | Path") -> Optional[Any]:

255 """Load a ``.pa.py`` file and return the provider class it defines.

256

257 The file is executed inside the safe sandbox. The module is expected to

258 define a class named ``Provider``; if that name is absent the first class

259 with a ``create_completion`` or ``create_async_generator`` attribute is

260 returned instead.

261

262 Args:

263 file_path: Path to the ``.pa.py`` file.

264

265 Returns:

266 The provider class, or ``None`` if none could be found.

267

268 Raises:

269 FileNotFoundError: If *file_path* does not exist.

270 ValueError: If *file_path* does not end with ``.pa.py``.

271 RuntimeError: If the file fails to execute.

272 """

273 file_path = Path(file_path)

274 if not file_path.exists():

275 raise FileNotFoundError(f"PA provider file not found: {file_path}")

276 if not file_path.name.endswith(".pa.py"):

277 raise ValueError(f"File must have .pa.py extension: {file_path}")

278

279 code = file_path.read_text(encoding="utf-8")

280 result = execute_safe_code(code)

281

282 if not result.success:

283 raise RuntimeError(

284 f"Failed to load PA provider from {file_path}:\n{result.error}"

285 )

286

287 # Prefer an explicit 'Provider' name

288 provider_class = result.locals.get("Provider")

289 if provider_class is not None:

290 return provider_class

291

292 # Fall back to any class that looks like a provider

293 for obj in result.locals.values():

294 if isinstance(obj, type) and (

295 hasattr(obj, "create_completion") or hasattr(obj, "create_async_generator")

296 ):

297 return obj

298

299 return None

300

301

302 def list_pa_providers(directory: "Optional[str | Path]" = None) -> List[Path]:

303 """Return all ``.pa.py`` files found (recursively) in *directory*.

304

305 Args:

306 directory: Directory to search. Defaults to the workspace.

307

308 Returns:

309 Sorted list of :class:`pathlib.Path` objects.

310 """

311 if directory is None:

312 directory = get_workspace_dir()

313 directory = Path(directory)

314 if not directory.exists():

315 return []

316 return sorted(directory.rglob("*.pa.py"))

Modified g4f/mcp/server.py +14 -4

@@ -26,8 +26,10 @@ from ..cookies import read_cookie_files

26 26 from ..image import EXTENSIONS_MAP

27 27 from ..image.copy_images import get_media_dir, copy_media, get_source_url

28 28

29 from .tools import MarkItDownTool, TextToAudioTool, WebSearchTool, WebScrapeTool, ImageGenerationTool

30 from .tools import WebSearchTool, WebScrapeTool, ImageGenerationTool

29 from .tools import (

30 MarkItDownTool, TextToAudioTool, WebSearchTool, WebScrapeTool, ImageGenerationTool,

31 PythonExecuteTool, FileReadTool, FileWriteTool, FileListTool, FileDeleteTool,

32 )

31 33

32 34

33 35 @dataclass

@@ -63,12 +65,20 @@ class MCPServer:

63 65 'web_scrape': WebScrapeTool(),

64 66 'image_generation': ImageGenerationTool(),

65 67 'text_to_audio': TextToAudioTool(),

66 'mark_it_down': MarkItDownTool()

68 'mark_it_down': MarkItDownTool(),

69 'python_execute': PythonExecuteTool(),

70 'file_read': FileReadTool(),

71 'file_write': FileWriteTool(),

72 'file_list': FileListTool(),

73 'file_delete': FileDeleteTool(),

67 74 }

68 75 self.server_info = {

69 76 "name": "gpt4free-mcp-server",

70 77 "version": "1.0.0",

71 "description": "MCP server providing web search, scraping, and image generation capabilities"

78 "description": (

79 "MCP server providing web search, scraping, image generation, "

80 "safe Python execution, and workspace file management capabilities"

81 ),

72 82 }

73 83

74 84 def get_tool_list(self) -> List[Dict[str, Any]]:

Modified g4f/mcp/tools.py +299 -1

@@ -4,11 +4,18 @@ This module provides MCP tool implementations that wrap gpt4free capabilities:

4 4 - WebSearchTool: Web search using ddg search

5 5 - WebScrapeTool: Web page scraping and content extraction

6 6 - ImageGenerationTool: Image generation using various AI providers

7 - PythonExecuteTool: Safe Python code execution with whitelisted modules

8 - FileReadTool: Read files from the ~/.g4f/workspace directory

9 - FileWriteTool: Write files to the ~/.g4f/workspace directory

10 - FileListTool: List files in the ~/.g4f/workspace directory

11 - FileDeleteTool: Delete files from the ~/.g4f/workspace directory

7 12 """

8 13

9 14 from __future__ import annotations

10 15

11 from typing import Any, Dict

16 import os

17 from pathlib import Path

18 from typing import Any, Dict, List

12 19 from abc import ABC, abstractmethod

13 20 import urllib.parse

14 21

@@ -459,3 +466,294 @@ class TextToAudioTool(MCPTool):

459 466 return {

460 467 "error": f"Text-to-speech URL generation failed: {str(e)}"

461 468 }

469

470

471 class PythonExecuteTool(MCPTool):

472 """Safe Python code execution tool with whitelisted module imports.

473

474 Executes the supplied code snippet inside a restricted sandbox where only

475 a curated list of modules may be imported and file-system access is limited

476 to the ``~/.g4f/workspace`` directory. The value assigned to the ``result``

477 variable (if any) is returned along with captured stdout/stderr.

478 """

479

480 @property

481 def description(self) -> str:

482 return (

483 "Execute a Python code snippet safely. Only whitelisted modules may "

484 "be imported (math, json, re, datetime, asyncio, aiohttp, g4f, …). "

485 "File access is restricted to the ~/.g4f/workspace directory. "

486 "Assign the value you want back to a variable named 'result'. "

487 "Returns stdout, stderr, and the value of 'result'."

488 )

489

490 @property

491 def input_schema(self) -> Dict[str, Any]:

492 return {

493 "type": "object",

494 "properties": {

495 "code": {

496 "type": "string",

497 "description": "Python code to execute in the safe sandbox",

498 },

499 "allowed_extra_modules": {

500 "type": "array",

501 "items": {"type": "string"},

502 "description": (

503 "Optional list of additional module names to allow "

504 "beyond the default whitelist"

505 ),

506 },

507 },

508 "required": ["code"],

509 }

510

511 async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:

512 from .pa_provider import execute_safe_code, SAFE_MODULES

513

514 code = arguments.get("code", "")

515 if not code:

516 return {"error": "code parameter is required"}

517

518 extra_names = arguments.get("allowed_extra_modules") or []

519 allowed = SAFE_MODULES | frozenset(extra_names)

520

521 try:

522 exec_result = execute_safe_code(code, allowed_modules=allowed)

523 return exec_result.to_dict()

524 except Exception as exc:

525 return {"error": f"Execution error: {exc}"}

526

527

528 class FileReadTool(MCPTool):

529 """Read a file from the ``~/.g4f/workspace`` directory."""

530

531 @property

532 def description(self) -> str:

533 return (

534 "Read the text content of a file inside the ~/.g4f/workspace directory. "

535 "Provide a relative path from the workspace root."

536 )

537

538 @property

539 def input_schema(self) -> Dict[str, Any]:

540 return {

541 "type": "object",

542 "properties": {

543 "path": {

544 "type": "string",

545 "description": "Relative path to the file inside the workspace",

546 }

547 },

548 "required": ["path"],

549 }

550

551 async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:

552 from .pa_provider import get_workspace_dir

553

554 rel_path = arguments.get("path", "")

555 if not rel_path:

556 return {"error": "path parameter is required"}

557

558 workspace = get_workspace_dir()

559 try:

560 target = (workspace / rel_path).resolve()

561 if not str(target).startswith(str(workspace.resolve())):

562 return {"error": "Access outside the workspace is not allowed"}

563 if not target.exists():

564 return {"error": f"File not found: {rel_path}"}

565 if not target.is_file():

566 return {"error": f"Path is not a file: {rel_path}"}

567 content = target.read_text(encoding="utf-8")

568 return {

569 "path": rel_path,

570 "content": content,

571 "size": len(content),

572 }

573 except Exception as exc:

574 return {"error": f"Read failed: {exc}"}

575

576

577 class FileWriteTool(MCPTool):

578 """Write (or create) a file inside the ``~/.g4f/workspace`` directory."""

579

580 @property

581 def description(self) -> str:

582 return (

583 "Write text content to a file inside the ~/.g4f/workspace directory. "

584 "Creates parent directories as needed. "

585 "Provide a relative path from the workspace root and the content to write."

586 )

587

588 @property

589 def input_schema(self) -> Dict[str, Any]:

590 return {

591 "type": "object",

592 "properties": {

593 "path": {

594 "type": "string",

595 "description": "Relative path to the file inside the workspace",

596 },

597 "content": {

598 "type": "string",

599 "description": "Text content to write to the file",

600 },

601 "append": {

602 "type": "boolean",

603 "description": "If true, append to existing file instead of overwriting (default: false)",

604 "default": False,

605 },

606 },

607 "required": ["path", "content"],

608 }

609

610 async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:

611 from .pa_provider import get_workspace_dir

612

613 rel_path = arguments.get("path", "")

614 content = arguments.get("content")

615 append = bool(arguments.get("append", False))

616

617 if not rel_path:

618 return {"error": "path parameter is required"}

619 if content is None:

620 return {"error": "content parameter is required"}

621

622 workspace = get_workspace_dir()

623 try:

624 target = (workspace / rel_path).resolve()

625 if not str(target).startswith(str(workspace.resolve())):

626 return {"error": "Access outside the workspace is not allowed"}

627 target.parent.mkdir(parents=True, exist_ok=True)

628 if append:

629 with open(target, "a", encoding="utf-8") as f:

630 f.write(content)

631 else:

632 target.write_text(content, encoding="utf-8")

633 return {

634 "path": rel_path,

635 "size": len(content),

636 "appended": append,

637 }

638 except Exception as exc:

639 return {"error": f"Write failed: {exc}"}

640

641

642 class FileListTool(MCPTool):

643 """List files and directories inside the ``~/.g4f/workspace`` directory."""

644

645 @property

646 def description(self) -> str:

647 return (

648 "List files and directories inside the ~/.g4f/workspace directory. "

649 "Optionally provide a subdirectory path relative to the workspace root. "

650 "Returns names, types, and sizes."

651 )

652

653 @property

654 def input_schema(self) -> Dict[str, Any]:

655 return {

656 "type": "object",

657 "properties": {

658 "path": {

659 "type": "string",

660 "description": (

661 "Relative path to a subdirectory inside the workspace "

662 "(default: workspace root)"

663 ),

664 "default": "",

665 },

666 "recursive": {

667 "type": "boolean",

668 "description": "If true, list files recursively (default: false)",

669 "default": False,

670 },

671 },

672 "required": [],

673 }

674

675 async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:

676 from .pa_provider import get_workspace_dir

677

678 rel_path = arguments.get("path", "") or ""

679 recursive = bool(arguments.get("recursive", False))

680

681 workspace = get_workspace_dir()

682 try:

683 target = (workspace / rel_path).resolve() if rel_path else workspace.resolve()

684 if not str(target).startswith(str(workspace.resolve())):

685 return {"error": "Access outside the workspace is not allowed"}

686 if not target.exists():

687 return {"error": f"Directory not found: {rel_path or '/'}"}

688 if not target.is_dir():

689 return {"error": f"Path is not a directory: {rel_path}"}

690

691 entries = []

692 iterator = target.rglob("*") if recursive else target.iterdir()

693 for entry in sorted(iterator):

694 try:

695 rel = str(entry.relative_to(workspace))

696 info: Dict[str, Any] = {

697 "path": rel,

698 "type": "file" if entry.is_file() else "directory",

699 }

700 if entry.is_file():

701 info["size"] = entry.stat().st_size

702 entries.append(info)

703 except Exception:

704 continue

705

706 return {

707 "workspace": str(workspace),

708 "path": rel_path or "/",

709 "entries": entries,

710 "count": len(entries),

711 }

712 except Exception as exc:

713 return {"error": f"List failed: {exc}"}

714

715

716 class FileDeleteTool(MCPTool):

717 """Delete a file from the ``~/.g4f/workspace`` directory."""

718

719 @property

720 def description(self) -> str:

721 return (

722 "Delete a file from the ~/.g4f/workspace directory. "

723 "Provide a relative path from the workspace root. "

724 "Only files can be deleted; directories are not removed."

725 )

726

727 @property

728 def input_schema(self) -> Dict[str, Any]:

729 return {

730 "type": "object",

731 "properties": {

732 "path": {

733 "type": "string",

734 "description": "Relative path to the file inside the workspace",

735 }

736 },

737 "required": ["path"],

738 }

739

740 async def execute(self, arguments: Dict[str, Any]) -> Dict[str, Any]:

741 from .pa_provider import get_workspace_dir

742

743 rel_path = arguments.get("path", "")

744 if not rel_path:

745 return {"error": "path parameter is required"}

746

747 workspace = get_workspace_dir()

748 try:

749 target = (workspace / rel_path).resolve()

750 if not str(target).startswith(str(workspace.resolve())):

751 return {"error": "Access outside the workspace is not allowed"}

752 if not target.exists():

753 return {"error": f"File not found: {rel_path}"}

754 if not target.is_file():

755 return {"error": f"Path is not a file (directories cannot be deleted): {rel_path}"}

756 target.unlink()

757 return {"path": rel_path, "deleted": True}

758 except Exception as exc:

759 return {"error": f"Delete failed: {exc}"}