(self, code_str, params, name, description, tool_id)
| 301 | return "\n".join(code_parts) |
| 302 | |
| 303 | def generate_mcp_server_code(self, code_str, params, name, description, tool_id): |
| 304 | code = self._generate_mcp_server_code(code_str, params, name, description, tool_id) |
| 305 | set_run_user = ( |
| 306 | f"os.setgid({pwd.getpwnam(_run_user).pw_gid});os.setuid({pwd.getpwnam(_run_user).pw_uid});" |
| 307 | if _enable_sandbox |
| 308 | else "" |
| 309 | ) |
| 310 | return f""" |
| 311 | import os, sys, logging |
| 312 | logging.basicConfig(level=logging.WARNING) |
| 313 | logging.getLogger("mcp").setLevel(logging.ERROR) |
| 314 | logging.getLogger("mcp.server").setLevel(logging.ERROR) |
| 315 | path_to_exclude = ['/opt/py3/lib/python3.11/site-packages', '/opt/maxkb-app/apps'] |
| 316 | sys.path = [p for p in sys.path if p not in path_to_exclude] |
| 317 | sys.path += {_sandbox_python_sys_path} |
| 318 | {set_run_user} |
| 319 | os.environ.clear() |
| 320 | exec({dedent(code)!a}) |
| 321 | """ |
| 322 | |
| 323 | def get_tool_mcp_config(self, tool, params): |
| 324 | _code = self.generate_mcp_server_code(tool.code, params, tool.name, tool.desc, str(tool.id)) |
no test coverage detected