Treasure Data MCP Server

Instructions

Implementation Reference

• td_mcp_server/mcp_impl.py:512-589 (handler)

  The core handler function for the 'td_read_project_file' tool. It validates the archive and file paths for security, opens the tar.gz archive, safely extracts the specified file member, reads its content with size limits, decodes as UTF-8 or Latin-1, and returns the content along with metadata. Includes comprehensive error handling.
  @mcp.tool() async def td_read_project_file(archive_path: str, file_path: str) -> dict[str, Any]: """Read the contents of a specific file from a project archive. This tool extracts and reads a specific file from a project archive, returning its contents. This allows examining SQL queries, workflow definitions, and other files without fully extracting the archive. Args: archive_path: The path to the downloaded project archive (.tar.gz file) file_path: The path of the file within the archive to read """ # Input validation - prevent path traversal if not _validate_archive_path(archive_path): return _format_error_response("Invalid archive path") if not _validate_file_path(file_path): return _format_error_response("Invalid file path") try: if not os.path.exists(archive_path): return _format_error_response("Archive file not found") try: with tarfile.open(archive_path, "r:gz") as tar: try: file_info = tar.getmember(file_path) # Security check for the member if not _safe_extract_member(file_info, "/tmp/validation"): return _format_error_response( "File access denied for security reasons" ) # Don't try to read directories if file_info.isdir(): return _format_error_response("Cannot read directory contents") # Extract and read the file f = tar.extractfile(file_info) if f is None: return _format_error_response("Failed to extract file") # Read with size limit if file_info.size > MAX_READ_SIZE: return _format_error_response("File too large to read") content_bytes = f.read() # Try to decode as text try: content = content_bytes.decode("utf-8") except UnicodeDecodeError: try: content = content_bytes.decode("latin-1") except UnicodeDecodeError: return _format_error_response( "File is not readable as text" ) extension = Path(file_path).suffix.lower() return { "success": True, "file_path": file_path, "content": content, "size": file_info.size, "extension": extension, } except KeyError: return _format_error_response("File not found in archive") except tarfile.ReadError: return _format_error_response("Invalid or corrupted archive file") except (OSError, UnicodeDecodeError) as e: return _format_error_response(f"Failed to read file: {str(e)}") except Exception as e: return _format_error_response(f"Unexpected error while reading file: {str(e)}")