(self, instance, with_valid=True)
| 466 | |
| 467 | @transaction.atomic |
| 468 | def insert(self, instance, with_valid=True): |
| 469 | if with_valid: |
| 470 | self.is_valid(raise_exception=True) |
| 471 | ToolCreateRequest(data=instance).is_valid(raise_exception=True) |
| 472 | # 校验代码是否包括禁止的关键字 |
| 473 | if instance.get("tool_type") == ToolType.MCP: |
| 474 | ToolExecutor().validate_mcp_transport(instance.get("code", "")) |
| 475 | |
| 476 | # 处理 work_flow_template |
| 477 | if instance.get("work_flow_template") is not None: |
| 478 | template_instance = instance.get("work_flow_template") |
| 479 | download_url = template_instance.get("downloadUrl") |
| 480 | if not download_url.startswith("https://apps-assets.fit2cloud.com/"): |
| 481 | raise AppApiException(500, _("Illegal download url")) |
| 482 | # 查找匹配的版本名称 |
| 483 | res = requests.get(download_url, timeout=5) |
| 484 | tool = ToolSerializer.Import( |
| 485 | data={ |
| 486 | "file": bytes_to_uploaded_file(res.content, "file.tool"), |
| 487 | "user_id": self.data.get("user_id"), |
| 488 | "workspace_id": self.data.get("workspace_id"), |
| 489 | "folder_id": str(instance.get("folder_id", self.data.get("workspace_id"))), |
| 490 | } |
| 491 | ).import_(name=instance.get("name"), source="template") |
| 492 | |
| 493 | try: |
| 494 | download_callback_url = template_instance.get("downloadCallbackUrl", "") |
| 495 | if not download_callback_url.startswith("https://apps.fit2cloud.com"): |
| 496 | raise AppApiException(500, _("Illegal download callback url")) |
| 497 | requests.get(download_callback_url, timeout=5) |
| 498 | except Exception as e: |
| 499 | maxkb_logger.error(f"callback appstore tool download error: {e}") |
| 500 | return tool |
| 501 | |
| 502 | tool_id = uuid.uuid7() |
| 503 | Tool( |
| 504 | id=tool_id, |
| 505 | name=instance.get("name"), |
| 506 | desc=instance.get("desc"), |
| 507 | code=instance.get("code"), |
| 508 | user_id=self.data.get("user_id"), |
| 509 | workspace_id=self.data.get("workspace_id"), |
| 510 | input_field_list=instance.get("input_field_list", []), |
| 511 | init_field_list=instance.get("init_field_list", []), |
| 512 | scope=instance.get("scope", ToolScope.WORKSPACE), |
| 513 | tool_type=instance.get("tool_type", ToolType.CUSTOM), |
| 514 | folder_id=instance.get("folder_id", self.data.get("workspace_id")), |
| 515 | is_active=False, |
| 516 | ).save() |
| 517 | |
| 518 | # 自动授权给创建者 |
| 519 | UserResourcePermissionSerializer( |
| 520 | data={ |
| 521 | "workspace_id": self.data.get("workspace_id"), |
| 522 | "user_id": self.data.get("user_id"), |
| 523 | "auth_target_type": AuthTargetType.TOOL.value, |
| 524 | } |
| 525 | ).auth_resource(str(tool_id)) |
no test coverage detected