from __future__ import annotations import base64 import json import mimetypes import random import string from pathlib import Path from typing import Any from urllib.parse import quote from mcp.server.fastmcp import FastMCP from ..config import DEFAULT_PROFILE from ..errors import QingflowApiError, raise_tool_error from ..json_types import JSONObject from .base import ToolBase ATTACHMENT_UPLOAD_INFO_FALLBACK_CODES = {"40118", 40118} LEGACY_OSS_FORM_REQUIRED_KEYS = ("key", "policy", "signature", "ossAccessKeyId") class FileTools(ToolBase): def register(self, mcp: FastMCP) -> None: @mcp.tool() def file_get_upload_info( profile: str = DEFAULT_PROFILE, upload_kind: str = "attachment", file_name: str = "", file_size: int = 0, upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict[str, Any]: return self.file_get_upload_info( profile=profile, upload_kind=upload_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) @mcp.tool() def file_upload_local( profile: str = DEFAULT_PROFILE, upload_kind: str = "attachment", file_path: str = "", upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict[str, Any]: return self.file_upload_local( profile=profile, upload_kind=upload_kind, file_path=file_path, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) def file_get_upload_info( self, *, profile: str, upload_kind: str, file_name: str, file_size: int, upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict[str, Any]: def runner(session_profile, context): upload_info = self._request_upload_info_with_fallback( context, upload_kind=upload_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) return { "profile": profile, "ws_id": session_profile.selected_ws_id, "upload_kind": upload_kind, "effective_upload_kind": upload_info["effective_upload_kind"], "upload_fallback_applied": upload_info["fallback_applied"], "file_name": file_name, "file_size": file_size, "request_route": self.backend.describe_route(context), "result": upload_info["result"], } return self._run(profile, runner) def file_upload_local( self, *, profile: str, upload_kind: str, file_path: str, upload_mark: str | None = None, content_type: str | None = None, bucket_type: str | None = None, path_id: int | None = None, file_related_url: str | None = None, ) -> dict[str, Any]: path = Path(file_path).expanduser() if not path.is_file(): raise_tool_error(QingflowApiError.config_error("file_path must point to an existing file")) file_name = path.name file_size = path.stat().st_size resolved_content_type = content_type or mimetypes.guess_type(file_name)[0] or "application/octet-stream" def runner(session_profile, context): upload_info = self._request_upload_info_with_fallback( context, upload_kind=upload_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=resolved_content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) result = upload_info["result"] if not isinstance(result, dict): raise QingflowApiError.config_error("upload endpoint did not return a structured upload payload") content = path.read_bytes() upload_protocol = "binary_put" if self._is_legacy_oss_form_upload(result): upload_result = self._upload_legacy_oss_form( result, file_name=file_name, content=content, content_type=resolved_content_type, ) upload_protocol = "oss_form_post" else: upload_url = str(result.get("uploadUrl") or "").strip() if not upload_url: raise QingflowApiError.config_error("upload endpoint did not return uploadUrl") upload_result = self.backend.upload_binary(upload_url, content, content_type=resolved_content_type) download_url = result.get("downloadUrl") return { "profile": profile, "ws_id": session_profile.selected_ws_id, "upload_kind": upload_kind, "effective_upload_kind": upload_info["effective_upload_kind"], "upload_fallback_applied": upload_info["fallback_applied"], "file_name": file_name, "file_size": file_size, "content_type": resolved_content_type, "request_route": self.backend.describe_route(context), "upload_protocol": upload_protocol, "result": result, "upload_result": upload_result, "download_url": download_url, "attachment_value": { "value": download_url, "name": file_name, }, "comment_file_info": { "url": download_url, "name": file_name, "uploadFileSize": file_size, }, } return self._run(profile, runner) def _build_file_upload_bo( self, *, upload_kind: str, file_name: str, file_size: int, upload_mark: str | None, content_type: str | None, bucket_type: str | None, path_id: int | None, file_related_url: str | None, ) -> dict[str, Any]: if not file_name: raise_tool_error(QingflowApiError.config_error("file_name is required")) if file_size <= 0: raise_tool_error(QingflowApiError.config_error("file_size must be positive")) if upload_kind == "attachment" and not upload_mark: raise_tool_error(QingflowApiError.config_error("upload_mark is required for attachment uploads and should usually be the app_key")) payload: dict[str, Any] = { "fileName": file_name, "fileSize": file_size, } if upload_mark: payload["uploadMark"] = upload_mark if content_type: payload["contentType"] = content_type if bucket_type: payload["bucketType"] = bucket_type if path_id is not None: payload["pathId"] = path_id if file_related_url: payload["fileRelatedUrl"] = file_related_url return payload def _resolve_upload_endpoint(self, upload_kind: str) -> str: normalized = upload_kind.strip().lower() if normalized == "attachment": return "/upload/puburl" if normalized == "login": return "/upload/url" if normalized == "anonymous": return "/upload/anonymousurl" raise_tool_error(QingflowApiError.config_error("upload_kind must be one of: attachment, login, anonymous")) raise AssertionError("unreachable") def _encode_formula(self, formula: str) -> str: encoded = base64.b64encode(quote(formula, safe="").encode("utf-8")).decode("utf-8") return f"{self._random_string(16)}{encoded}{self._random_string(16)}" def _random_string(self, length: int) -> str: alphabet = string.ascii_letters + string.digits return "".join(random.choice(alphabet) for _ in range(length)) def _request_upload_info_with_fallback( self, context, *, upload_kind: str, file_name: str, file_size: int, upload_mark: str | None, content_type: str | None, bucket_type: str | None, path_id: int | None, file_related_url: str | None, ) -> dict[str, Any]: requested_kind = upload_kind.strip().lower() attempted_kinds = [requested_kind] if requested_kind == "attachment": attempted_kinds.append("login") last_error: QingflowApiError | None = None for current_kind in attempted_kinds: try: result = self._request_upload_info( context, upload_kind=current_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) return { "requested_upload_kind": requested_kind, "effective_upload_kind": current_kind, "fallback_applied": current_kind != requested_kind, "result": result, } except QingflowApiError as error: last_error = error if not self._should_retry_attachment_upload_info(requested_kind, current_kind, error): raise assert last_error is not None raise last_error def _request_upload_info( self, context, *, upload_kind: str, file_name: str, file_size: int, upload_mark: str | None, content_type: str | None, bucket_type: str | None, path_id: int | None, file_related_url: str | None, ) -> JSONObject: endpoint = self._resolve_upload_endpoint(upload_kind) file_upload_bo = self._build_file_upload_bo( upload_kind=upload_kind, file_name=file_name, file_size=file_size, upload_mark=upload_mark, content_type=content_type, bucket_type=bucket_type, path_id=path_id, file_related_url=file_related_url, ) encrypted_payload = { "upload": self._encode_formula(json.dumps(file_upload_bo, ensure_ascii=False, separators=(",", ":"))), "fileName": file_name, "fileSize": file_size, } if path_id is not None: encrypted_payload["pathId"] = path_id if bucket_type: encrypted_payload["bucketType"] = bucket_type result = self.backend.request("POST", context, endpoint, json_body=encrypted_payload) if not isinstance(result, dict): raise QingflowApiError.config_error("upload endpoint did not return a structured upload payload") return result def _should_retry_attachment_upload_info( self, requested_kind: str, attempted_kind: str, error: QingflowApiError, ) -> bool: return ( requested_kind == "attachment" and attempted_kind == "attachment" and error.backend_code in ATTACHMENT_UPLOAD_INFO_FALLBACK_CODES ) def _is_legacy_oss_form_upload(self, payload: JSONObject) -> bool: return all(str(payload.get(key) or "").strip() for key in LEGACY_OSS_FORM_REQUIRED_KEYS) def _upload_legacy_oss_form( self, upload_info: JSONObject, *, file_name: str, content: bytes, content_type: str, ) -> dict[str, Any]: upload_url = self._resolve_legacy_oss_form_url(upload_info) if not upload_url: raise QingflowApiError.config_error("legacy upload payload is missing host/uploadAccelerateHost") form_fields = { "key": str(upload_info["key"]), "policy": str(upload_info["policy"]), "signature": str(upload_info["signature"]), "OSSAccessKeyId": str(upload_info["ossAccessKeyId"]), } callback = str(upload_info.get("callback") or "").strip() if callback: form_fields["callback"] = callback return self.backend.upload_form_file( upload_url, form_fields=form_fields, file_field="file", file_name=file_name, content=content, content_type=content_type, ) def _resolve_legacy_oss_form_url(self, upload_info: JSONObject) -> str: for key in ("uploadAccelerateHost", "host", "uploadUrl"): value = str(upload_info.get(key) or "").strip() if not value: continue if value.startswith(("http://", "https://")): return value return f"https://{value.lstrip('/')}" return ""